Prefetch и честное распределение

Два одинаковых консьюмера, очередь из тысячи задач. Первый в мыле, второй скучает. Виноват не код обработки, а одна пропущенная строчка.

prefetch_count (задаётся через basic_qos) — сколько неподтверждённых сообщений брокер разрешает держать консьюмеру одновременно. Пока лимит выбран, следующее сообщение уйдёт кому-то другому.

По умолчанию RabbitMQ раздаёт сообщения round-robin — по кругу, не глядя, кто чем занят. И это разумно: брокер не знает и не может знать, сколько времени займёт обработка. Пока задачи одинаковые, круговая раздача — это ровно то, что нужно. Но стоит появиться разнобою (одна задача — отчёт на десять секунд, другая — пинг на пятьдесят миллисекунд), как раздача превращается в лотерею, и лотерею вы регулярно проигрываете.

Симуляция: сколько стоит round-robin

Смоделируем на чистом Python — восемь задач, чередующихся тяжёлая/лёгкая, два консьюмера. Слева честная раздача «по кругу», справа — «следующему освободившемуся».

import heapq

tasks = [10, 1, 10, 1, 10, 1, 10, 1]   # секунд на обработку каждой задачи

# 1. Round-robin: брокер раздал заранее, по очереди, не глядя на загрузку
workers = [0, 0]
for i, cost in enumerate(tasks):
    workers[i % 2] += cost
print("round-robin: A =", workers[0], "сек, B =", workers[1], "сек, всего =", max(workers), "сек")

# 2. prefetch_count=1: сообщение уходит тому, кто уже освободился
free = [(0, "A"), (0, "B")]            # (когда освободится, кто)
load = {"A": 0, "B": 0}
for cost in tasks:
    ready_at, who = heapq.heappop(free)
    load[who] += cost
    heapq.heappush(free, (ready_at + cost, who))
finish = max(moment for moment, _ in free)
print("prefetch=1:  A =", load["A"], "сек, B =", load["B"], "сек, всего =", finish, "сек")

Результат:

round-robin: A = 40 сек, B = 4 сек, всего = 40 сек
prefetch=1:  A = 22 сек, B = 22 сек, всего = 22 сек

Работы в обоих случаях одинаково — 44 секунды. Но при круговой раздаче все тяжёлые задачи достались консьюмеру A: он молотит 40 секунд, а B заканчивает на четвёртой и 36 секунд смотрит в потолок. Простое «не давай мне следующее, пока я не подтвердил текущее» даёт 22 секунды — идеальную утилизацию. Купить второй сервер здесь было бы бесполезно: узкое место не в мощности, а в раздаче.

Как включить честную раздачу

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
channel.queue_declare(queue="tasks", durable=True)

# ДО basic_consume: «не давай мне второе сообщение, пока я не подтвердил первое»
channel.basic_qos(prefetch_count=1)

def on_message(ch, method, properties, body):
    process(body)                                   # может занять и 50 мс, и 10 секунд
    ch.basic_ack(delivery_tag=method.delivery_tag)  # только теперь просим следующее

channel.basic_consume(queue="tasks", on_message_callback=on_message)
channel.start_consuming()

Три условия, без любого из которых basic_qos превращается в декорацию:

1. basic_qos вызван до basic_consume. Иначе брокер уже начал слать сообщения по старым правилам, и первая пачка уедет мимо лимита.
2. Подтверждения ручные. С auto_ack=True prefetch просто игнорируется: неподтверждённых сообщений не существует в принципе, счётчик всегда ноль, лимит недостижим — брокер шлёт без остановки. Это самая частая история «поставил prefetch, ничего не изменилось».
3. Ack действительно вызывается. Забыли подтвердить в ветке с ошибкой — счётчик unacked не уменьшился. При prefetch_count=1 консьюмер после первой же такой ошибки замолкает навсегда: процесс жив, лог пуст, сообщения копятся.

На консьюмера или на канал

channel.basic_qos(prefetch_count=10, global_qos=False)  # по 10 каждому консьюмеру канала (умолчание)
channel.basic_qos(prefetch_count=10, global_qos=True)   # 10 на весь канал, как бы его ни делили

В pika параметр называется global_qos — global в Python зарезервировано. Трактовка у RabbitMQ своя, отличная от буквы спецификации AMQP 0-9-1: global=False — лимит на каждого консьюмера, global=True — на канал целиком. Если в одном канале живут несколько консьюмеров и вы хотите ограничить суммарное потребление памяти — берите global=True.

Какое значение ставить

prefetch_count=1 — это не «лучшая практика», а один из полюсов. Он максимизирует справедливость и минимизирует пропускную способность.

Эмпирика для старта: prefetch ≈ (время обработки + round-trip до брокера) / время обработки, округлив вверх. Это минимальное значение, при котором консьюмер не простаивает в ожидании следующего сообщения. Обработка 5 мс, RTT 1 мс — хватит 2. Обработка 200 мкс, RTT 1 мс — нужно 6 и больше. Дальше меряйте: смотрите messages_unacknowledged. Если оно стабильно равно prefetch × число консьюмеров, вы упёрлись в скорость обработки, и поднимать prefetch бессмысленно — нужны консьюмеры.

Как это работает

Брокер ведёт счётчик неподтверждённых сообщений на каждого консьюмера (при global=False). Отправил сообщение — плюс один, получил ack — минус один. Как только счётчик достиг prefetch_count, консьюмер для брокера «занят», и следующее сообщение уходит любому другому, у кого есть свободный слот. Если свободных нет — сообщение просто остаётся в очереди в состоянии Ready и ждёт.

Отсюда и берётся справедливость: раздача идёт не «по кругу», а «тому, кто попросил». Фактически prefetch_count=1 превращает push-протокол в pull-модель — консьюмер сам определяет свой темп, а ack становится запросом на следующую порцию работы.

Из той же механики следуют важные ограничения:

• Prefetch не делает медленного консьюмера быстрым. Он лишь не даёт грузить его сверх меры. Если обработка не успевает за потоком, очередь всё равно будет расти — нужны дополнительные консьюмеры.
• Prefetch — это не параллелизм. prefetch_count=10 в однопоточном консьюмере не значит «обрабатываем десять сразу»: девять сообщений просто лежат в буфере клиента и ждут. Параллелизм даёт пул потоков/процессов или несколько отдельных консьюмеров.
• Prefetched-сообщения не бесплатны. Они висят в unacked у брокера и в памяти клиента. При падении консьюмера все они возвращаются в очередь с флагом redelivered — то есть будут обработаны повторно.

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

rabbitmqctl list_consumers queue_name consumer_tag prefetch_count

Результат:

queue_name  consumer_tag       prefetch_count
tasks       ctag1.9a3f4c...    1
tasks       ctag2.b71ce9...    1

Ноль в колонке prefetch_count означает «без ограничений» — то есть basic_qos не сработал (или консьюмер подписан с auto_ack), и никакого честного распределения у вас нет.

Частые ошибки

• basic_qos после basic_consume. Брокер уже начал слать — лимит применится с опозданием или не применится вовсе.
• basic_qos вместе с auto_ack=True. Лимит игнорируется полностью. Prefetch работает только на ручных подтверждениях.
• prefetch_count=1 «потому что так в туториале» на потоке из десятков тысяч коротких сообщений в секунду. Пропускная способность падает в разы, а профита ноль: задачи и так одинаковые, распределять нечего.
• Не подтвердить сообщение при исключении. При prefetch_count=1 консьюмер после этого мёртв как рабочая единица, оставаясь живым как процесс. Отлаживать такое без метрик мучительно.
• Огромный prefetch при долгой обработке. Сообщения застревают у одного консьюмера: новые подключившиеся воркеры простаивают, а при падении «богатого» консьюмера в очередь разом падает сотня redelivered-сообщений.
• Один и тот же prefetch на разных по мощности машинах. Лимит настраивается на консьюмере, а не «в конфиге на всех» — пользуйтесь этим.
• Ждать, что prefetch решит проблему приоритетов. Не решит: он выравнивает загрузку, но не меняет порядок. Нужны приоритеты — заведите отдельную очередь для тяжёлых задач и отдельный пул консьюмеров. Это проще и предсказуемее, чем x-max-priority.

Итоги

• По умолчанию RabbitMQ раздаёт сообщения по кругу, не зная о загрузке консьюмеров. На неравномерных задачах это приводит к тому, что один захлёбывается, а второй простаивает.
• channel.basic_qos(prefetch_count=N) ограничивает число неподтверждённых сообщений у консьюмера — и превращает раздачу в «тому, кто освободился».
• Работает только в связке с ручным ack и только если вызван до basic_consume.
• prefetch_count=1 — максимальная справедливость, минимальная пропускная способность. Для коротких задач берите десятки.
• Отправная точка: prefetch ≈ (время обработки + RTT) / время обработки, дальше — по метрике messages_unacknowledged.
• Prefetch управляет раздачей, а не скоростью и не параллелизмом: медленный консьюмер быстрее не станет.