Есть поток идентификаторов. Для каждого нужно выполнить поиск, который занимает
не меньше 500 мс, и переложить результат в отдельный канал. Последовательно 11 id
обрабатываются 5.5 с, а нужно примерно 0.5 с.
type found struct {
id int
}
func main() {
ids := make(chan int)
go func() {
defer close(ids)
for i := 0; i <= 10; i++ {
ids <- i
}
}()
// TODO: результаты поиска по каждому id — в отдельный канал
}
func lookup(id int) found {
time.Sleep(500 * time.Millisecond)
return found{id: id}
}
Оформи стадию конвейера функцией. lookup передаётся параметром, чтобы его можно
было подменить:
type Found struct {
ID int
}
// Process вызывает lookup для каждого id из ids, не больше чем в workers горутинах
// одновременно (workers < 1 считается как 1), и отдаёт результаты в возвращаемый канал.
func Process(ctx context.Context, ids <-chan int, workers int, lookup func(id int) Found) <-chan Found
Требования
- Для каждого id из
ids вызывается ровно один lookup, и его результат
попадает в выходной канал. Порядок результатов любой.
- Одновременно выполняется не больше
workers вызовов lookup, и все
workers действительно работают параллельно: 20 id по 100 мс при
workers = 20 обрабатываются примерно за 100 мс.
- Выходной канал закрывается, когда
ids закрыт и все результаты отданы.
Process не блокируется и сразу возвращает канал.
- При отмене
ctx выходной канал закрывается, а горутины Process
завершаются (после того, как вернутся уже начатые lookup), даже если
потребитель перестал читать или ids никогда не закроют.
- Код проходит
go test -race.
Пример
out := Process(ctx, ids, 11, lookup)
for f := range out {
fmt.Println(f.ID) // все 0..10 в каком-то порядке, через ≈500 мс
}
// ProcessOrdered — как Process, но результаты выходят в том же порядке, в каком id пришли из ids.
func ProcessOrdered(ctx context.Context, ids <-chan int, workers int, lookup func(id int) Found) <-chan Found
- Порядок на выходе совпадает с порядком входа, даже если поздние
lookup
завершаются раньше ранних.
- Параллельность та же: не больше
workers одновременных lookup, и все они
реально работают одновременно.
- Закрытие, отмена и отсутствие утечек — как у
Process.
Устно: где в конвейере узкое место, если потребитель читает медленнее, чем
работают воркеры? Как прокинуть отмену внутрь lookup? Как обрабатывать ошибку
поиска: отдельное поле в результате, отдельный канал ошибок или отмена всего
конвейера?