Воркер-пулы и конвейеры

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

Пул воркеров — самая частая конкурентная конструкция в Go-бэкенде: фиксированный набор горутин разбирает поток задач, вместо того чтобы запускать горутину на каждую. Базовый пул — двадцать строк. Но на практике к нему почти всегда приходят дополнительные требования: пережить панику в задаче, обслуживать срочные задачи раньше, выдавать результаты по порядку, не превышать лимит запросов в секунду, остановиться за заданное время, расти и сжиматься под нагрузкой. Каждая задача блока добавляет одно такое требование.

Общая тема — жизненный цикл воркеров и задач: как запустить, как распределить работу, как остановиться без потерь и утечек. Остановка пула — это несколько разных действий (перестать принимать, доделать принятое, отменить выполняющееся, дождаться выхода), и каждая задача точно говорит, какие из них нужны.

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

• Фиксированный пул: канал задач, WaitGroup, закрытие канала результатов координатором.
• recover вокруг одной задачи, а не вокруг всего воркера.
• Приоритетные очереди под мьютексом и sync.Cond.
• Сохранение порядка: нумерация или канал каналов.
• Один общий лимитер частоты на весь пул.
• Контекст у каждой задачи и остановка с таймаутом.
• Pub/sub: доставка каждому подписчику, а не одному из воркеров.
• Счётчики живых и занятых воркеров для масштабирования.

Задачи и порядок решения

Решайте от фиксированного пула к самым сложным вариантам: 24 → 28 → 29 → 30 → 27 → 31 → 25 → 26 → 23 → 32.

• 24 · Panic-Safe Worker Pool — паника в задаче превращается в ошибку, воркер продолжает работу.
• 28 · Context-Aware Worker Pool — у каждой задачи свой ctx; отменённая задача сама завершается, и воркер берёт следующую.
• 29 · Fan-Out с маршрутизацией по тегам (Pub/Sub) — сообщение получают только подписчики его тега; при остановке очередь дренируется.
• 30 · Graceful Shutdown с таймаутом — Stop перестаёт принимать задачи, ждёт выполняющиеся не дольше timeout и возвращает ID не успевших.
• 27 · Rate-Limited Pipeline (50 RPS) — двухстадийный конвейер с общим лимитом частоты на весь пул.
• 31 · Cron Worker Pool — периодические задачи со своими интервалами на общем пуле, без горутины на каждый тик.
• 25 · Priority Worker Pool — три уровня приоритета, воркеры берут высший доступный, Submit не блокируется.
• 26 · Ordered Worker Pool — обработка параллельная, выдача строго в порядке поступления.
• 23 · Worker Pool с динамическим масштабированием — число воркеров растёт под очередь и сжимается при простое, в границах minWorkers/maxWorkers.
• 32 · Idle Timeout Pool — minWorkers постоянных, рост до maxWorkers, выход лишних после idleTimeout.

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

• Паттерны конкурентности — пул, семафор, errgroup и раздел «Перед задачами блоков 5 и 7»: приоритеты, порядок, масштабирование, остановка, паника, pub/sub.
• sync: примитивы — WaitGroup, мьютексы и Cond для состояния пула.
• context — отмена задач и остановка с таймаутом.

// Идиома блока: воркер читает задачи и выходит, когда канал закрыт.
func (p *Pool) worker() {
    defer p.wg.Done()
    for task := range p.tasks {
        task()
    }
}