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

Отдельные каналы и мьютексы — это буквы. Паттерны конкурентности — это слова, из которых собираются реальные пайплайны обработки данных. Fan-in, fan-out, bounded worker pool, многостадийный конвейер, батчинг — именно из этих кирпичей строятся загрузчики логов, обработчики очередей, ETL-пайплайны и агрегаторы. Этот блок превращает понимание примитивов в умение проектировать поток данных: где распараллелить, где собрать обратно, как ограничить параллелизм и как сгруппировать мелкие операции в эффективные пачки.

Сквозная тема — владение каналами и закрытием. Кто создаёт канал, тот его и закрывает; закрытие распространяется вниз по конвейеру и аккуратно завершает все стадии без утечек.

Что вы научитесь делать

• Сливать произвольное число каналов в один (fan-in) с корректным закрытием выхода по sync.WaitGroup.
• Ограничивать число одновременных воркеров (bounded pool) и собирать ошибки, не теряя их.
• Строить многостадийный конвейер, где у каждой стадии свой пул горутин, а данные текут через цепочку каналов.
• Группировать события в батчи по размеру или по времени — паттерн, снижающий нагрузку на БД и внешние API.
• Видеть, как закрытие входного канала каскадом завершает весь пайплайн.

Карта задач

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

Связанные главы учебника

• Паттерны конкурентности — pipeline, fan-in/fan-out, worker pool, semaphore, batching.
• Каналы — владение каналом, закрытие вниз по конвейеру, направления как контракт между стадиями.

// Идиома блока: 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) }()