Prefetch и fair dispatch: QoS потребителя

Сценарий, который случается в каждом втором проекте с очередями: два воркера слушают одну очередь, в ней копятся задачи, и вот первый воркер получил пять тяжёлых выгрузок подряд, а второй пьёт чай. Round-robin работает — сообщения раздаются по кругу, — но брокер раздаёт не по скорости обработки, а по факту доставки: отправил и забыл. Причина в том, что без ограничений брокер заполняет буфер каждого потребителя, пока тот физически принимает данные, а воркеры-то подтвердили бы готовность — да не спросили. Решение называется prefetch: потребитель объявляет, сколько сообщений готов держать неподтверждёнными одновременно, и брокер ставит рассылку на паузу, когда окно заполнено. Первый воркер замолчал после первой задачи — очередь переключается на второго. Один метод, один аргумент, и распределение перестаёт зависеть от порядка доставки. В этом уроке — сигнатура, выбор числа, цена ошибки и проверка на живой очереди.

Qos rabbitmq: окно неподтверждённых доставок#

Сигнатура ch.Qos(prefetchCount, prefetchSize int, global bool) error задаёт настройки потребления канала, и разбор qos rabbitmq стоит начать с главного аргумента. PrefetchCount — сколько доставок брокер может держать неподтверждёнными для потребителя: получил две, подтвердил одну — окно открылось на одно, и поедет следующее; окно заполнилось — брокер ждёт ack, и это не тайм-аут, а штатная пауза. PrefetchSize — лимит в байтах на неподтверждённые сообщения: серверы RabbitMQ его не реализуют, передавайте ноль, значение больше нуля не даст эффекта. Global определяет область действия: false применяет настройку к каждому потребителю канала отдельно — так живут почти все воркеры; true делает окно общим на всех потребителей канала — вариант редкий, полезен, когда на одном канале живёт несколько подписок и надо ограничить их суммарный аппетит. Важное правило порядка: Qos вызывают до ConsumeWithContext — настройка влияет на потребителей, зарегистрированных после вызова, а на уже подписанных не действует. Ошибка возвращается редко: канал закрыт или число отрицательное.

От round-robin к fair dispatch#

Почему prefetch меняет поведение брокера, лучше видно на механике доставки. RabbitMQ толкает сообщения: воркеру не нужно спрашивать — брокер сам пишет в его поток, пока есть что и куда. Без лимита «куда» определяется только сетевым буфером, поэтому быстрый сетевой воркер собирает хвост очереди у себя, даже если обрабатывает со скоростью улитки. Prefetch добавляет обратную связь: пока неподтверждённых доставок меньше лимита, брокер шлёт новые; достигли лимита — этот потребитель выпадает из рассылки до ближайшего ack. Теперь round-robin считается по реальной готовности: медленный воркер заморозил окно — его очередь минуют; быстрый подтвердил — получил следующее. Это и называется fair dispatch, честное распределение. Связь с предыдущим уроком прямая: окно считается только по неподтверждённым доставкам, то есть по manual ack; с autoAck=true сообщение снимается с учёта в момент доставки, окно пустое, prefetch не работает вовсе. Прежде чем настраивать честность, нужно убедиться, что потребитель вообще подтверждает вручную.

Воркер с окном на три задачи

package main

import (
        "context"
        "log"
        "time"

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

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

        ch, err := conn.Channel()
        if err != nil {
                log.Fatal(err)
        }

        if err := ch.Qos(3, 0, false); err != nil {
                log.Fatal(err)
        }

        _, err = ch.QueueDeclare("tasks", true, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }

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

        msgs, err := ch.ConsumeWithContext(ctx, "tasks", "fair-worker",
                false, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }

        for d := range msgs {
                log.Println("начал:", string(d.Body))
                time.Sleep(300 * time.Millisecond)
                if err := d.Ack(false); err != nil {
                        log.Println("ack failed:", err)
                }
                log.Println("готово:", string(d.Body))
        }
}

Qos вызывается до подписки — с окном в три неподтверждённые доставки. Пока воркер держит три неподтверждённых, четвёртое сообщение остаётся в очереди; каждый Ack открывает окно на одно место. Запустите два таких воркера и гоните продюсером из шестого урока смесь коротких и длинных задач: при prefetch равном единице брокер перестанет скармливать всё первому.

Какой prefetch выбрать#

Значение	Поведение	Когда уместно
0	окно без лимита — рассылка ограничена только сетью	дефолт, который почти всегда надо поменять
1	строгий fair dispatch: одна задача — один воркер	тяжёлые задачи с непредсказуемой длительностью
2-10	компромисс: скорость против честности	типовые воркеры с задачами в десятки миллисекунд
десятки и больше	честность исчезает, растёт память и цена рестарта	обоснованная партийная обработка с батч-ack
global=true	одно окно на всех потребителей канала	несколько подписок с общим лимитом

Забытый Qos: очередь-пылесос

Кейс: воркер обработки чеков запустили без Qos — а продюсер в этот момент разогнался до тысяч сообщений в минуту. Брокер честно наполнил буфер потребителя: несколько тысяч сообщений улетели в один процесс, остальные воркеры стояли с пустыми руками, а в management UI глубина очереди падала, хотя работы оставалось на часы. Долгая очередь unacked у одного процесса — это ещё и риск: рестарт возвращает в очередь всё содержимое окна, а долгие доставки в хвосте окна упираются в consumer_timeout из восьмого урока. Диагноз виден на вкладке очереди: у одного потребителя тысячи unacked, у остальных ноль. Лечение — ch.Qos перед подпиской, и код-ревью, где подписка без установленного окна не проходит.

Tradeoff: скорость против честности#

У prefetch есть цена с обеих сторон, и это главный повод не копировать чужие числа. Маленькое окно делает честным, но добавляет сетевые рейсы: каждое сообщение требует доставки и ack, и при prefetch равном единице воркер половину времени может ждать кадры по сети, особенно когда брокер в другом дата-центре; на медленных задачах это незаметно, на быстрых — режет пропускную способность. Большое окно восстанавливает несправедливость: воркер копит неподтверждённый хвост, память растёт на буфере, рестарт возвращает всё окно в очередь, и простаивающие соседи отгребают повторную работу. Рабочая настройка выводится из времени обработки: прикиньте, сколько задач воркер успевает сделать за время прихода следующей, — это и есть естественная глубина окна; для быстрых задач с батч-подтверждением из восьмого урока окно логично связать с размером партии. Универсального числа нет: пятьдесят миллисекунд и пятьдесят секунд на задачу требуют разного prefetch, и настраивают его измерением, а не догмой.

Окно из конфигурации

prefetch := 3
if v := os.Getenv("WORKER_PREFETCH"); v != "" {
        if n, err := strconv.Atoi(v); err == nil && n > 0 {
                prefetch = n
        }
}

if err := ch.Qos(prefetch, 0, false); err != nil {
        log.Fatal(err)
}

msgs, err := ch.ConsumeWithContext(ctx, "tasks", "fair-worker",
        false, false, false, false, nil)

Число, которое приходится тюнить по нагрузке, не должно жить в константе кода: выносите prefetch в переменную окружения или конфиг и меняйте без пересборки. PrefetchSize передан нулём — лимита в байтах серверы RabbitMQ не реализуют, а global=false значит, что каждый новый потребитель канала получит своё окно.

Переполнение и мониторинг окна#

Настроенное окно живёт своей жизнью, и наблюдать за ним надо теми же цифрами, что и очередь. Пара «глубина очереди, unacked» рассказывает всё: unacked у воркеров равен prefetch — окно работает, воркеры заняты, добавьте процессов, если глубина растёт; unacked у всех ноль при растущей глубине — воркеры не тянут или окна слишком малы, работа стоит, а не распределена. Неправильное окно узнаётся по всплескам: воркер с огромным prefetch на пике держит тысячи неподтверждённых, и его рестарт превращается в лавину requeue, которая сама по себе тормозит очередь. Долгие доставки в глубине большого окна — ещё и путь к consumer_timeout из восьмого урока: сообщение ждёт ack тридцать минут и закрывает канал. На графике удобно вести обе величины через Gauge из курса по мониторингу: глубина отвечает «сколько работы накопилось», unacked — «сколько работы прямо сейчас у воркеров», и вместе они отвечают на вопрос масштабирования быстрее любого профилирования.

Как увидеть чужой prefetch: UI и rabbitmqctl#

Окно выставляет код воркера, а проверять его приходится на живой системе — и тут полезно знать, где брокер показывает настройки потребителей. В management UI на вкладке очереди раскройте секцию Consumers: у каждой подписки виден prefetch count, тег и канал; потребитель без лимита показывается нулём — это та самая бездонная рассылка. В консоли то же самое даёт rabbitmqctl list_consumers с колонкой prefetch_count: удобно, когда вкладки браузера нет, а подозрение на очередь-пылесос есть. Для скриптов и проверок в пайплайне годится HTTP API management-плагина: эндпоинт с потребителями отдаёт JSON с теми же полями, и скрипт может ругаться, найдя подписку с нулевым окном на боевой очереди. Так настройка из внутреннего соглашения превращается в проверяемый факт: код ставит окно, ревью его видит, а мониторинг ловит, если кто-то задеплоил воркер без Qos.

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

Prefetch виден только на живой очереди: в браузерной песочнице доступны лишь fmt, strings, strconv, math, sort, errors и time, без сети и брокера. Поднимите два воркера из урока против контейнера из второго, гоните нагрузку продюсером и сравните распределение при окнах один и десять; интерактивы ниже проверят сигнатуру, порядок вызовов и поведение autoAck.

Правила prefetch#

Qos вызывают до ConsumeWithContext: настройка действует на потребителей, подписанных после неё.
Prefetch работает только с ручными ack; autoAck обнуляет всю механику окна.
PrefetchSize передавайте нулём: лимит в байтах серверы RabbitMQ не реализуют.
Один — для тяжёлых задач, единицы — для быстрых; число выводят из времени обработки, а не копируют.
Нулевое окно — безлимит: оставлять его стоит только осознанно.
Мониторьте пару «глубина очереди, unacked»: она подсказывает, когда пора масштабировать воркеров.

Окружение этого урока: седьмой урок дал подписку и цикл воркера, восьмой — подтверждения, без которых окно не работает. Воркеры, выросшие из этого кода, собираются в семнадцатом уроке про worker pool, остановка пула без потерь — в двадцать первом про graceful shutdown, а очередь-пылесос остаётся в коллекции двадцать третьего урока про антипаттерны. Метрику unacked удобно вести через Gauge из курса по Prometheus. Дальше по курсу начинается маршрутизация: одиннадцатый урок выводит публикации за пределы дефолтного обменника и показывает direct exchange. Карта курса — на странице курса.

Проверка: механика окна

Воркер с autoAck=false вызвал ch.Qos(2, 0, false). Сколько неподтверждённых доставок может держать брокер для этого потребителя?

Проверка: prefetch и autoAck

Потребитель подписался с autoAck=true, а перед этим воркер вызвал ch.Qos(1, 0, false). Что произойдёт?

Что напечатает воркер?

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

ch.Qos(1, 0, false)
msgs, _ := ch.ConsumeWithContext(ctx, "tasks", "w",
        false, false, false, false, nil)

for d := range msgs {
        fmt.Println("начал:", string(d.Body))
        // обработка зависла на первой задаче, ack не вызывается
        time.Sleep(time.Hour)
}

Соберите настройку окна

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

err := ch.(3, 0, false) if err != nil { log.Fatal(err) } msgs, err := ch.(ctx, "tasks", "fair-worker", false, false, false, false, nil) for d := range msgs { d.(false) }

Соберите воркер с fair dispatch

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

Строки

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

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

Карточки: prefetch и Qos

Термин

ch.Qos

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

Определение

Настройка потребления канала: prefetchCount, prefetchSize, global. Вызывается до подписки и действует на потребителей, подписанных после.

1 / 6

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

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

Вывод

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

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

ch.Qos(prefetchCount, 0, false) до подписки — базовая настройка окна для воркера с ручными ack.
Окно считается по неподтверждённым доставкам: заполнено — брокер ставит рассылку до ближайшего ack.
Fair dispatch — это round-robin по реальной готовности: занятый воркер пропускается.
autoAck обнуляет механику окна; prefetchSize серверы RabbitMQ не реализуют — ноль.
Маленькое окно честно, но тратит рейсы на сеть; большое восстанавливает несправедливость, память и лавину requeue.
Число выводят из времени обработки и нагрузки, выносят в конфиг и меняют без пересборки.
Пара «глубина очереди, unacked» — главный график масштабирования воркеров.

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

Запустите два воркера с окном один, гоните нагрузку продюсером из шестого урока со смесью задач разной длительности и убедитесь, что занятый воркер больше не получает задачи, пока не подтвердит. Затем поднимите окно до двадцати и сравните распределение unacked в management UI. В конце вынесите prefetch в переменную окружения, как в примере, и проверьте, что окно меняется без пересборки воркера.

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

Какой prefetch ставить, если задачи разные по длительности?

Для смеси лучше низкое окно — единица или двойка: тяжёлая задача перестаёт блокировать честность, а редкие рейсы сети на фоне секунд обработки незаметны. Высокие окна имеет смысл поднимать под однородные быстрые задачи и батч-подтверждения, где экономия сетевых рейсов заметна.

Почему Qos вызван, но воркер всё равно получает сообщения без ограничений?

Проверьте две вещи: вызван ли Qos до ConsumeWithContext — настройка действует только на подписки после него, и не подписан ли потребитель с autoAck=true — при автоматических подтверждениях окно пустое и лимит игнорируется. Третий редкий случай: Qos на другом канале — настройки канала не переносятся.

Что лучше: одно окно global=true или отдельное на потребителя?

Практически всегда отдельное, global=false: окно на потребителя даёт fair dispatch между воркерами. Global=true полезен в специфичной схеме, когда несколько подписок живут на одном канале и нужно ограничить их суммарный аппетит — например, чтобы не собрать всё unacked в одном процессе.

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

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