Как реализовать worker pool с использованием каналов в Go

Что проверяют

Умеете ли вы правильно закрыть каналы и не получить взаимную блокировку.

Как отвечать

Схема: канал задач, фиксированное число горутин-рабочих, канал результатов.

func Run(ctx context.Context, jobs []Job, workers int) []Result {
    in := make(chan Job)
    out := make(chan Result)

    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for job := range in {
                select {
                case out <- process(job):
                case <-ctx.Done():
                    return
                }
            }
        }()
    }

    go func() {
        defer close(in)
        for _, j := range jobs {
            select {
            case in <- j:
            case <-ctx.Done():
                return
            }
        }
    }()

    go func() {
        wg.Wait()
        close(out)
    }()

    var res []Result
    for r := range out {
        res = append(res, r)
    }
    return res
}

Что здесь принципиально и что чаще всего ломают:

• канал задач закрывает отправитель после отправки всех задач. Именно закрытие завершает циклы range у рабочих;
• канал результатов закрывается в отдельной горутине после ожидания группы. Закрыть его в основном потоке нельзя: он в это время читает из него, и получилась бы взаимная блокировка. Закрыть его из рабочего тоже нельзя, рабочих несколько;
• чтение результатов идёт параллельно с работой. Если сначала дождаться всех, а потом читать, небуферизованный канал заблокирует рабочих на первой же отправке;
• контекст в обеих ветках: и при отправке, и при возврате результата. Иначе при отмене горутина навсегда зависнет на записи в канал, который никто не читает.

Размер пула подбирают по характеру задач: для счётных ориентируются на число ядер, для сетевых значение может быть в разы больше, потому что рабочие в основном ждут.

Стоит упомянуть errgroup: он даёт ту же схему с ограничением одновременности и сбором первой ошибки, и в реальном коде обычно берут его.

Где спотыкаются

Закрывают канал результатов из рабочего.

Ждут завершения всех рабочих до начала чтения и получают взаимную блокировку.

Не слушают контекст при отправке в канал.

Забывают закрыть канал задач, и рабочие висят вечно.