--- title: "Prefetch и fair dispatch: QoS потребителя" description: "Prefetch в amqp091-go: метод Qos, честное распределение задач между воркерами, tradeoff окна неподтверждённых и переполнение. Qos rabbitmq в Go с примерами." course: "RabbitMQ в Go: очереди сообщений на amqp091-go" url: "https://golangmaster.ru/courses/rabbitmq/lessons/10-prefetch/" published: "2026-09-19" updated: "2026-09-21" minutes: 18 --- # Prefetch и fair dispatch: QoS потребителя *Курс «RabbitMQ в Go: очереди сообщений на amqp091-go», урок 10 из 25. ~18 минут. Обновлено 21.09.2026. Источник: https://golangmaster.ru/courses/rabbitmq/lessons/10-prefetch/* Честную раздачу задач между воркерами делает qos rabbitmq: prefetch ограничивает окно неподтверждённых доставок и не даёт брокеру завалить одного воркера, пока другой простаивает. Разберём ch.Qos по аргументам, посмотрим, как prefetch превращает механический round-robin в fair dispatch, где проходит tradeoff между честностью и пропускной способностью и что видит дежурный, когда окно настроено неправильно. Сценарий, который случается в каждом втором проекте с очередями: два воркера слушают одну очередь, в ней копятся задачи, и вот первый воркер получил пять тяжёлых выгрузок подряд, а второй пьёт чай. 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 не работает вовсе. Прежде чем настраивать честность, нужно убедиться, что потребитель вообще подтверждает вручную. ### Воркер с окном на три задачи ```go 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, и настраивают его измерением, а не догмой. ### Окно из конфигурации ```go 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»: она подсказывает, когда пора масштабировать воркеров. Окружение этого урока: [седьмой урок](/courses/rabbitmq/lessons/07-potreblenie-soobschenij/) дал подписку и цикл воркера, [восьмой](/courses/rabbitmq/lessons/08-ack-nack/) — подтверждения, без которых окно не работает. Воркеры, выросшие из этого кода, собираются в [семнадцатом уроке про worker pool](/courses/rabbitmq/lessons/17-worker-pool/), остановка пула без потерь — в [двадцать первом про graceful shutdown](/courses/rabbitmq/lessons/21-graceful-rabbitmq/), а очередь-пылесос остаётся в коллекции [двадцать третьего урока про антипаттерны](/courses/rabbitmq/lessons/23-antipatterny-ocheredey/). Метрику unacked удобно вести через [Gauge из курса по Prometheus](/courses/prometheus/lessons/05-gauge/). Дальше по курсу начинается маршрутизация: [одиннадцатый урок](/courses/rabbitmq/lessons/11-obmenniki-direct/) выводит публикации за пределы дефолтного обменника и показывает direct exchange. Карта курса — на [странице курса](/courses/rabbitmq/). ### Проверка: механика окна **Вопрос:** Воркер с autoAck=false вызвал ch.Qos(2, 0, false). Сколько неподтверждённых доставок может держать брокер для этого потребителя? - ✅ Две: окно открывается на новое сообщение после каждого ack - Один: prefetch=2 означает одну задачу в обработке - Без ограничений: глобальный лимит задаёт только сервер - Ноль: prefetch работает только с autoAck=true PrefetchCount задаёт лимит неподтверждённых доставок на потребителя: два выдали — рассылка встала, один ack — окно открылось на одно место. Наоборот неверно: prefetch имеет смысл только при autoAck=false. ### Проверка: prefetch и autoAck **Вопрос:** Потребитель подписался с autoAck=true, а перед этим воркер вызвал ch.Qos(1, 0, false). Что произойдёт? - Брокер ограничит рассылку одним сообщением - ✅ Prefetch игнорируется: autoAck снимает доставку с учёта, окно пустое - Брокер вернёт ошибку на подписку - Первое сообщение придёт с задержкой на один ack Окно считает неподтверждённые доставки, а autoAck подтверждает их в момент отправки — считать нечего, ограничение не работает. Это самая частая связка-ловушка: fair dispatch начинается с manual ack из восьмого урока. ### Что напечатает воркер? ```go 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) } ``` **Что выведет программа?** - ✅ Одну строку «начал» — окно заполнено, брокер молчит до ack - Все строки подряд: рассылка не зависит от окна - Ни одной: Qos после подписки не действует - Две строки: окно всегда пропускает с запасом Qos вызван до подписки, окно равно единице: первая доставка заняла место, ack нет — брокер ставит рассылку этому потребителю. Именно так выглядит штатная пауза окна, а не зависание сети: очередь в этот момент делит сообщения между другими воркерами. ### Соберите настройку окна Допишите установку Qos, подписку и подтверждение в цикле воркера. ```go 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) } ``` Подсказка: Окно ставит ch.Qos до подписки, подписку открывает ConsumeWithContext, место в окне освобождает Ack. Правильные ответы в пропусках: `Qos`, `ConsumeWithContext`, `Ack` ### Соберите воркер с fair dispatch Расставьте строки в порядке выполнения: соединение, канал, окно, подписка, чтение, подтверждение. Правильный порядок строк: 1. `conn, err := amqp.Dial(url)` 2. `ch, err := conn.Channel()` 3. `err = ch.Qos(3, 0, false)` 4. `msgs, err := ch.ConsumeWithContext(ctx, "tasks", "", false, false, false, false, nil)` 5. `d := <-msgs` 6. `d.Ack(false)` ### Карточки: prefetch и Qos - **ch.Qos** — Настройка потребления канала: prefetchCount, prefetchSize, global. Вызывается до подписки и действует на потребителей, подписанных после. - **prefetchCount** — Лимит неподтверждённых доставок на потребителя: окно заполнилось — рассылка встаёт до ближайшего ack. - **prefetchSize** — Лимит окна в байтах: серверами RabbitMQ не реализован, передают ноль. - **global** — Область окна: false — каждому потребителю канала своё, true — одно окно на всех потребителей канала. - **fair dispatch** — Честное распределение: занятый воркер с заполненным окном пропускается, задачи уходят свободным. Требует ручных ack. - **prefetch и autoAck** — Несовместимы: autoAck подтверждает доставку в момент отправки, окно пустое, лимит не работает. ## Коротко: что запомнить - 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 в одном процессе.