Нужен конвейер, который обрабатывает элементы параллельно в workers горутинах, а результаты отдаёт строго в том порядке, в каком получил входы. Если элемент 5 посчитался раньше элемента 2, он всё равно ждёт своей очереди.
// ProcessOrdered читает in, применяет work к каждому элементу в workers
// горутинах и пишет результаты в возвращаемый канал в порядке поступления.
// Возвращаемый канал закрывается, когда обработан весь вход.
func ProcessOrdered(in <-chan int, workers int, work func(int) int) <-chan int
Вход 0, 1, 2, 3, 4, функция work удваивает число, но на чётных засыпает дольше, чем на нечётных. Нечётные посчитаются раньше, а в выходе всё равно:
in := make(chan int)
go func() {
for i := 0; i < 5; i++ {
in <- i
}
close(in)
}()
work := func(x int) int {
if x%2 == 0 {
time.Sleep(50 * time.Millisecond)
}
return x * 2
}
for v := range ProcessOrdered(in, 3, work) {
fmt.Println(v)
}
// 0
// 2
// 4
// 6
// 8
- Результаты в
out идут строго в порядке входа: i-й прочитанный элемент даёт i-й результат.
- Обработка параллельная: пока медленный элемент считается, соседние не ждут.
out закрывается ровно один раз после того, как выданы все результаты. Пустой вход даёт сразу закрытый out.- Для
workers < 1 ведите себя как для одного воркера. Тесты проверяют workers = 1, workers = 10, а также 0 и отрицательные значения (там work не должен выполняться параллельно).
- Код проходит
go test -race, горутины после закрытия out не остаются (тест считает их через runtime.NumGoroutine).
Рассчитывайте на штатный сценарий: вход рано или поздно закрывается, work возвращается, а потребитель дочитывает out до конца. Досрочная остановка чтения (отмена) в задачу не входит.
Тесты также проверяют, что work выполняется не больше чем в workers горутинах одновременно и что при workers = 8 реально идёт параллельная обработка (одновременно хотя бы четыре вызова).