Паттерн worker pool на очереди

Цикл воркера из седьмого урока читает доставки по одной, и на задачах в десятки миллисекунд этого хватает надолго. Но стоит обработке потянуть сеть, базу или картинку, как один потребитель становится узким местом: очередь растёт, а процессор процесса простаивает в ожидании ответов. Первая реакция — поднять prefetch, но окно из десятого урока ускоряет только раздачу, а не обработку: воркер по-прежнему один и делает задачи последовательно. Нужна конкурентность внутри потребителя — и это классический worker pool: несколько горутин выполняют одну функцию обработки над одним потоком задач. Очередь здесь играет роль разделителя: издатели пишут в неё, не зная, сколько воркеров читает, а пул масштабируется без правки издателей. Собрать пул на amqp091-go несложно, сложнее собрать правильно: с каналами AMQP по одному на горутину, честной раздачей через Qos и остановкой без потери недоделанных задач.

Rabbitmq worker: две схемы пула#

Для rabbitmq worker существуют две схемы, и выбор определяет весь остальной код. Первая: каждая горутина сама открывает канал и подписывается ConsumeWithContext — у пула нет диспетчера, брокер сам раздаёт сообщения по окнам prefetch. Вторая: одна горутина читает Delivery из единственной подписки и раскладывает задачи в Go-канал, из которого их разбирают воркеры. Схема с диспетчером выглядит привычнее для тех, кто вырос на worker pool из курса по Go, но добавляет лишний узел: одна подписка остаётся единственным потоком, справедливость между воркерами — его забота, а не брокера, и раздача упирается в одну горутину. Схема без диспетчера проще: брокер делает fair dispatch из десятого урока, каждый воркер подтверждает свои доставки на своём канале, а очередь держит излишки работы. Плата — N каналов вместо одного: по четвёртому уроку канал лёгкий, десятки каналов для брокера незаметны. Поэтому по умолчанию выбирают схему без диспетчера, а диспетчер оставляют для случаев, когда между чтением и воркером нужна дополнительная логика — маршрутизация по типу или батчирование.

Пул из четырёх воркеров

package main

import (
        "context"
        "log"
        "os"
        "strconv"
        "sync"

        amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
        workers := 4
        if v := os.Getenv("WORKERS"); v != "" {
                if n, err := strconv.Atoi(v); err == nil && n > 0 {
                        workers = n
                }
        }

        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        if err != nil {
                log.Fatal(err)
        }
        defer conn.Close()

        ctx, cancel := context.WithCancel(context.Background())
        defer cancel()

        var wg sync.WaitGroup
        for i := 1; i <= workers; i++ {
                ch, err := conn.Channel()
                if err != nil {
                        log.Fatal(err)
                }
                if err := ch.Qos(3, 0, false); err != nil {
                        log.Fatal(err)
                }
                tag := "worker-" + strconv.Itoa(i)
                msgs, err := ch.ConsumeWithContext(ctx, "tasks", tag,
                        false, false, false, false, nil)
                if err != nil {
                        log.Fatal(err)
                }
                wg.Add(1)
                go func() {
                        defer wg.Done()
                        defer ch.Close()
                        for d := range msgs {
                                log.Println(tag, "начал:", string(d.Body))
                                d.Ack(false)
                        }
                }()
        }

        wg.Wait() // здесь начинается дожим: ctx отменён — воркеры доедают
}

Каждая итерация цикла открывает свой канал, ставит окно Qos и подписывается с индивидуальным тегом: N горутин — N каналов, и ни одна доставка не путается с чужой. Отмена контекста закрывает потоки подписок: range завершается, воркер подтверждает последнее сообщение и выходит, а wg.Wait не даёт процессу умереть раньше дожима. Полный сценарий с сигналами и ack на выходе разобран в двадцать первом уроке.

Сколько воркеров поднимать#

Характер задач	Стартовое число	Ограничитель
процессорные	близко к GOMAXPROCS	лишние горутины только растят переключения
сетевые и ввод-вывод	десятки, по замеру латентности	пул соединений к базе и лимиты внешних API
тяжёлые с непредсказуемой длительностью	воркеров немного, prefetch=1	честность раздачи важнее пропускной способности
быстрые однотипные	воркеры плюс окно в единицы	сеть и размер очереди, а не число горутин

Один канал AMQP на всех воркеров

Кейс: команда взяла пул из туториала, но для экономии открыла один канал и разделила поток Delivery между горутинами. Сначала работало, потом начались необъяснимые ошибки подтверждений и обрывы канала под нагрузкой. Причина в контракте библиотеки: канал amqp091-go не рассчитан на одновременную работу нескольких горутин — ack, nack и публикации на одном канале из разных мест дают состояние гонки, которое race detector находит не сразу. Брокер в придачу закрывает канал на нарушение порядка кадров, и вместе с каналом умирают все его неподтверждённые доставки. Лечение одно и простое: канал на воркера, как в коде выше; общий допустим только для объявления топологии в одном месте. Если хочется полного диспетчера — берите схему с Go-каналом из предыдущего раздела, но подтверждение пусть делает та же горутина, которая подписалась.

Воркеры rabbitmq: Qos и теги#

Пул без настроенных окон повторяет грабли десятого урока: воркеры rabbitmq получают сообщения по мере сетевой готовности, и первые горутины собирают хвост очереди. Поэтому Qos ставится на каждый канал до подписки: окно в единицы превращает раздачу брокера в fair dispatch, а окно-единица подходит тяжёлым задачам. Тег потребителя — вторая мелочь, которая стоит дорого на дежурстве: individual tag вида worker-1 позволяет в management UI отличить подписки пула друг от друга, увидеть, кто держит unacked, и понять, какой воркер завис. Без тегов все потребители выглядят анонимным пустым тегом, и расследование превращается в угадывание. Prefetch умножается на количество воркеров: четыре воркера с окном три — это до двенадцати неподтверждённых доставок на процесс, и при выборе числа это стоит держать в голове вместе с рестартом: обрыв возвращает в очередь всё содержимое всех окон сразу.

Метрики пула: processed, failed, in-flight#

Пул без метрик — ящик без шкалы, и собрать шкалу проще всего атомарными счётчиками вокруг функции обработки. Processed и failed — Counter из четвёртого урока курса по Prometheus: прирост показывает живость пула, а доля ошибок — качество задач. In-flight — Gauge из пятого: плюс на входе в обработку, минус на выходе, и вместе с глубиной очереди из двадцать второго урока он отвечает на вопрос масштабирования: in-flight у потолка окон и растущая глубина — добавляйте воркеров или инстансов. Время обработки удобно мерить Histogram и по нему выбирать окно prefetch. Логи пишут с тегом воркера: по ним видно, что задачи распределяются, а не липнут к одному. Всё это — строки вокруг цикла, и начинать их стоит раньше, чем тюнить числа: без цифр любой выбор размера пула — гадание.

Воркер с метриками и recover

var processed, failed atomic.Int64
var inFlight atomic.Int64

func worker(tag string, msgs <-chan amqp.Delivery) {
        for d := range msgs {
                inFlight.Add(1)
                err := func() (err error) {
                        defer func() {
                                if r := recover(); r != nil {
                                        err = fmt.Errorf("паника: %v", r)
                                }
                        }()
                        return process(d.Body)
                }()
                inFlight.Add(-1)
                if err != nil {
                        failed.Add(1)
                        d.Nack(false, false) // в очередь повторов из урока 16
                        continue
                }
                processed.Add(1)
                d.Ack(false)
        }
}

Recover внутри замыкания превращает панику на одной задаче в обычную ошибку: без него паника убивает весь процесс вместе с остальными воркерами и неподтверждёнными доставками. Счётчики атомарные — их читает HTTP-эндпоинт метрик из другого потока; in-flight symmetric: плюс на входе, минус на выходе, и убывающий до нуля in-flight на дежим сигнализирует, что пул можно закрывать.

Пул в процессе или реплики?

Пул внутри процесса и несколько реплик процесса решают одну задачу разными способами, и они не взаимоисключающие. Внутренний пул дешевле: нет сетевых накладных на межпроцессную раздачу,(prefetch и окна работают между горутинами, добавление воркера — правка конфига. Реплики дают изоляцию: упавший инстанс не уносит работы соседей, обновление катится по частям, а горизонтальное масштабирование не упирается в один процесс. Типовая связка: пул из единиц горутин в каждом инстансе плюс две-три реплики; вместе они держат нагрузку лучше, чем один процесс с сотней горутин, потому что отказ домена меньше.

Дожим: WaitGroup и живой канал#

Остановка пула — то место, где теряют сообщения, поэтому механика важна в деталях. Отмена контекста, переданного в ConsumeWithContext, закрывает поток доставок: range завершается, воркер выходит. Сообщения, которые воркер уже получил, но не подтвердил, остаются unacked до его решения — значит, функция обработки обязана дойти до конца и подтвердить или отклонить. WaitGroup в main держит процесс живым, пока воркеры доедают: без него cancel, за которым следует выход из main, оборвал бы горутины на середине. Порядок в defer тоже имеет значение: conn.Close стоит после wg.Wait, потому что подтверждение на закрытом канале вернёт ErrClosed, и последнее сообщение уйдёт в очередь как непонятое. Если в очереди остались задачи, воркеры после отмены их не возьмут — поток подписки закрыт, и это правильно: остаток подберут воркеры после рестарта. Полный сценарий с сигналами и тайм-аутом дожима — в двадцать первом уроке, здесь важна схема: отмена — дожим — выход по WaitGroup.

Код урока — локально, интерактивы — на странице

Пул оживает только на брокере, поэтому полный код запускается локально против контейнера из второго урока: поднимите пул с WORKERS=4, погоните продюсер из шестого и посмотрите в management UI, как окна распределяются по тегам. В песочнице сети и пакета sync нет — там доступны fmt, strings, strconv, math, sort, errors и time, но саму механику «одна задача — один воркер» можно прочувствовать программой на Go-каналах и горутинах: две подписчицы читают из одного канала, каждая задача уходит ровно одной из них, а пауза вместо ожидания растянет гонку наглядно. Интерактивы ниже закрепят правила каналов и дожима.

Правила worker pool#

Свой канал AMQP на горутину: канал amqp091-go не для конкурентного использования из нескольких воркеров.
Qos до подписки на каждом канале: окна превращают раздачу в fair dispatch и защищают от очереди-пылесоса.
Индивидуальный тег потребителя на воркера: по нему дежурный отличает подписки в management UI.
autoAck=false обязателен: без ручных ack не работает ни окно, ни дожим.
Recover вокруг обработки: паника одной задачи не должна убивать процесс и чужие доставки.
Остановка по схеме: отмена контекста — дожим текущих задач — выход после wg.Wait, conn.Close в самом конце.

Соседи этого урока: седьмой урок про ConsumeWithContext дал цикл, который пул размножает, десятый про prefetch — окна и fair dispatch, а четвёртый про connection и channel — правило канала на горутину. Остановку с сигналами целиком разбирает двадцать первый урок про graceful shutdown, глубину очереди как сигнал масштабирования — двадцать второй про наблюдаемость. Горутины и WaitGroup остались в четвёртом и седьмом уроках курса по Go, метрики processed и in-flight — в курсе по Prometheus. Карта курса — на странице курса.

Проверка: каналы пула

Пул из восьми воркеров собран по схеме без диспетчера. Сколько каналов AMQP откроет процесс и почему?

Проверка: что случится при отмене

Контекст пула отменён, а один воркер как раз обрабатывает доставку — функция ещё не вернулась. Что произойдёт дальше?

Сколько строк напечатает пул?

Прочитайте код и выберите, что он выведет:

jobs := make(chan int, 3)
for i := 1; i <= 3; i++ {
        jobs <- i
}
close(jobs)

var wg sync.WaitGroup
for w := 1; w <= 2; w++ {
        wg.Add(1)
        go func(id int) {
                defer wg.Done()
                for j := range jobs {
                        fmt.Println("воркер", id, "взял", j)
                }
        }(w)
}
wg.Wait()

Соберите скелет пула

Допишите соединение, канал, окно и подтверждение в цикле воркера.

conn, err := amqp.(url) if err != nil { log.Fatal(err) } ch, err := conn.() if err := ch.(3, 0, false); err != nil { log.Fatal(err) } for d := range msgs { d.(false) }

Соберите запуск одного воркера

Расставьте строки в порядке выполнения: соединение, канал, окно, подписка, запуск горутины, подтверждение.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Карточки: worker pool

Термин

worker pool

нажмите, чтобы перевернуть

Определение

Несколько горутин обрабатывают поток задач из одной очереди. На amqp091-go собирается схемой без диспетчера: каждая горутина подписывается сама.

1 / 6

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Пул собирается схемой без диспетчера: каждая горутина открывает канал, ставит Qos и подписывается сама.
Канал amqp091-go не для конкурентного использования: один канал на всех воркеров — гонка и обрывы.
Окно на канал даёт fair dispatch; суммарный in-flight — окна, умноженные на число воркеров.
Индивидуальный consumer tag делает пул читаемым в management UI и логах.
Recover вокруг обработки превращает панику задачи в ошибку, а не в смерть процесса.
Дожим: отмена контекста закрывает поток, воркер доедает текущую задачу, WaitGroup держит процесс.
Размер пула выводят из характера задач и метрик processed, failed, in-flight — а не копируют из туториала.

Домашнее задание

Соберите пул из урока с числом воркеров из переменной окружения, поднимите четыре воркера с окном два и погоните нагрузку продюсером из шестого урока. В management UI у каждого воркера должен быть свой тег и своё окно, unacked распределяются по горутинам. Добавьте счётчики processed, failed и in-flight с выводом в лог по таймеру и убедитесь, что при отмене ctx in-flight убывает до нуля до выхода процесса. В конце сравните поведение при WORKERS=1 и WORKERS=8 на смеси быстрых и медленных задач.

Вопросы и ответы по уроку

Можно ли один ConsumeWithContext раздавать задачи в Go-канал?

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

Сколько воркеров ставить на старте?

Стартовая точка — единицы: четыре-восемь горутин с окном в единицы для смешанных задач, по GOMAXPROCS для процессорных и десятки для сетевых с быстрым внешним API. Дальше решают метрики: in-flight у потолка окон и растущая глубина очереди — сигнал поднять число, ошибки по лимитам зависимостей — сигнал уменьшить.

Нужен ли отдельный mutex на счётчики метрик?

Нет: atomic.Int64 и атомарные операции дешевле и проще мьютекса, а счётчики — инкременты и декременты. Мьютекс нужен, если метрика составная — например, скользящее среднее времени обработки; для простых счётчиков и Gauge атомики — идиоматичное решение, и race detector на них не ругается.

Термины урока

Ключевые понятия из этого урока — короткие определения в глоссарии. Наведите на термин, чтобы вспомнить, откройте карточку для подробностей и ссылки на практику.