Паттерны конкурентности

О чём этот блок

Загрузчик логов, обработчик очереди сообщений, ETL-конвейер, агрегатор метрик устроены одинаково: данные проходят через несколько стадий, некоторые стадии распараллелены, в конце результаты собираются вместе. Блок учит собирать такие конструкции из каналов и горутин: где распараллелить, где слить потоки обратно, как ограничить число одновременных воркеров и как сгруппировать мелкие операции в пачки.

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

Что понадобится

• Fan-in: слияние N каналов в один и закрытие выхода после wg.Wait().
• Ограниченный пул воркеров и сбор ошибок без гонок.
• Многостадийный конвейер, где у каждой стадии свой пул.
• Батчинг по двум условиям: набралось N элементов или вышло время.
• select с <-ctx.Done() рядом с каждой отправкой, если потребитель может уйти раньше.

Задачи

• 09 · Fan-In: Merge каналов — N входных каналов в один; выход закрывается ровно после завершения всех источников.
• 10 · Bounded Worker Pool с обработкой ошибок — не больше w воркеров одновременно, результаты и ошибки собираются без потерь.
• 11 · Многостадийный конвейер (Stage Pipeline) — Reader → Parser → Validator, у каждой стадии свой пул.
• 12 · Batcher (группировщик запросов) — сброс пачки по размеру или по таймеру, что наступит раньше.

Что прочитать

• Паттерны конкурентности — конвейер, fan-out/fan-in, пул, семафор, батчинг, errgroup.
• Каналы — владение закрытием и направления каналов как контракт между стадиями.

// Идиома блока: fan-in закрывает выход ровно после всех источников.
var wg sync.WaitGroup
for _, c := range channels {
    wg.Add(1)
    go func(c <-chan int) {
        defer wg.Done()
        for v := range c {
            out <- v
        }
    }(c)
}
go func() { wg.Wait(); close(out) }()