Подтверждения: ack и nack

Консьюмер упал посреди обработки: сообщение потеряно навсегда или вернётся в очередь? Ответ зависит от одного аргумента в вызове basic_consume.

Acknowledgement (ack, подтверждение) — сигнал консьюмера брокеру: «это сообщение я обработал, можешь его забыть». Пока ack не пришёл, RabbitMQ считает сообщение «в работе» и вернёт его в очередь, если консьюмер отвалится.

Типичная история. Сервис рассылки писем читает очередь emails, на каждое сообщение стучится в SMTP — это секунды. В момент деплоя контейнер получает SIGKILL. Наутро выясняется: сорок писем клиентам не ушло, а в очереди пусто. Логов нет, паники нет, брокер бодр и весел. Куда делись сообщения — вопрос, который придётся закрыть в первую же неделю работы с RabbitMQ.

Как теряются сообщения: auto_ack

Вот тот самый консьюмер. Найдите строчку, из-за которой письма исчезли.

import pika

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

def on_message(ch, method, properties, body):
    send_email(body)          # три секунды: SMTP, ретраи, всё как в жизни
    print("отправлено:", body)

# ОПАСНО: брокер удалит сообщение сразу, как только отдаст его в сокет
channel.basic_consume(queue="emails", on_message_callback=on_message, auto_ack=True)
channel.start_consuming()

Виновник — auto_ack=True (в протоколе AMQP этот режим называется no-ack, и это название честнее). Он означает «подтверждение по факту доставки»: брокер вычёркивает сообщение из очереди в тот момент, когда записывает его в TCP-сокет, — ещё до того, как ваш код его увидел. Дальше сообщение живёт только в памяти процесса-консьюмера. Упал процесс — не осталось ничего.

Такая семантика называется at-most-once: доставим не больше одного раза, а может быть, и ноль. Она законна и полезна ровно там, где потеря отдельного сообщения ничего не стоит: метрики, телеметрия, лог кликов, обновление «сколько человек сейчас на странице». Для писем, платежей и заказов это диверсия.

Второй, менее очевидный эффект: с auto_ack=True брокер не ограничивает поток. У него нет счётчика неподтверждённых сообщений (их просто не бывает), поэтому он вываливает консьюмеру всё, что есть в очереди, — сто тысяч сообщений улетают в клиентский буфер и съедают память. Один kill -9 — и очередь пуста, а работа не сделана.

Ручное подтверждение

Рабочий вариант выглядит так. Обратите внимание на порядок: сначала работа, потом ack.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()
channel.queue_declare(queue="emails", durable=True)
channel.basic_qos(prefetch_count=1)

def on_message(ch, method, properties, body):
    try:
        send_email(body)                                                # 1. сначала работа
    except SmtpTemporaryError:                                          # почта прилегла — повтор поможет
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)   # 2а. вернуть в очередь
    except SmtpAddressInvalid:                                          # адрес битый — повтор бесполезен
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)  # 2б. отдать в dead-letter
    else:
        ch.basic_ack(delivery_tag=method.delivery_tag)                  # 2в. только теперь «готово»

# auto_ack по умолчанию False — брокер будет ждать нашего ответа
channel.basic_consume(queue="emails", on_message_callback=on_message)
channel.start_consuming()

Разберём по строчкам.

basic_consume без auto_ack — это ручной режим. Брокер отдаёт сообщение и переводит его в состояние unacked: из очереди оно уже не выдаётся другим, но и не удалено.
method.delivery_tag — порядковый номер доставки в рамках канала (1, 2, 3…). Это не идентификатор сообщения: то же самое сообщение после повторной доставки получит другой tag, а в соседнем канале нумерация своя. Отсюда правило: подтверждать нужно тем же каналом, которым получили.
basic_ack вызывается строго после успешной работы. Формулировка «сначала подтвердим, чтобы не забыть» — это ровно auto_ack, только длиннее и с иллюзией контроля.
basic_nack(requeue=True) — «не смог, пусть попробует кто-нибудь ещё». Сообщение возвращается в очередь (RabbitMQ старается вернуть его на исходную позицию) и уходит любому свободному консьюмеру, в том числе снова вам. У повторной доставки поднят флаг method.redelivered.
basic_nack(requeue=False) — «это сообщение не починится». Оно уйдёт в dead-letter exchange, если он настроен у очереди, иначе будет удалено.
basic_reject — то же, что basic_nack, но умеет отклонять только одно сообщение. У nack есть multiple=True: подтвердить или отклонить пачку до текущего tag включительно. Это расширение RabbitMQ поверх AMQP, и именно поэтому в новом коде обычно пишут nack.

Ядовитое сообщение

Самая обидная авария на ручных подтверждениях выглядит так: в очередь попало сообщение с битым телом, обработчик падает, код честно делает basic_nack(requeue=True), сообщение возвращается в голову очереди, приходит обратно через миллисекунду, снова падает — и так тысячи раз в секунду. Это poison message, «ядовитое сообщение»: CPU в полке, лог забит одинаковой трассировкой, остальные сообщения стоят.

Лечится различением ошибок «повтор поможет» и «повтор не поможет» плюс страховкой на случай, если вы ошиблись:

import json

def on_message(ch, method, properties, body):
    try:
        payload = json.loads(body)
    except ValueError:                       # тело не парсится — повтор не поможет никогда
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        return

    try:
        handle(payload)
    except TemporaryError:
        # если это уже повторная доставка — карусель не крутим, отдаём в dead-letter
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=not method.redelivered)
        return

    ch.basic_ack(delivery_tag=method.delivery_tag)

Флаг redelivered — грубый инструмент: он говорит «это не первая доставка», но не говорит, какая по счёту. Точный счётчик попыток даёт связка с dead-letter exchange (заголовок x-death) — этим займёмся в разделе про DLX. На старте достаточно правила: бесконечный requeue запрещён.

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

У сообщения в очереди три состояния: Ready (лежит и ждёт), Unacked (отдано консьюмеру, ответа нет) и «удалено» (пришёл ack). Из Unacked есть два выхода назад в Ready: явный nack/reject с requeue=True — и обрыв.

Обрыв — ключевая часть механики. Unacked-сообщения привязаны к каналу. Закрылся канал, оборвался TCP-коннект, умер процесс консьюмера — брокер немедленно возвращает все его unacked-сообщения в очередь и помечает их redelivered. Никакого таймера ожидания нет, никакой «эвристики живости» — RabbitMQ узнаёт о смерти консьюмера из состояния сокета, поэтому восстановление занимает миллисекунды. Именно это и спасает те самые сорок писем.

Есть и обратная защита — от консьюмера, который жив, но завис. Начиная с версии 3.8.15 у брокера настроен consumer_timeout (по умолчанию 30 минут): если сообщение висит в unacked дольше, канал закрывается с ошибкой PRECONDITION_FAILED, а сообщение возвращается в очередь. Если у вас есть задачи, которые честно считаются час, — это ваш будущий инцидент: либо поднимайте таймаут в конфиге брокера, либо дробите работу на шаги.

Из ручного ack вытекает главное следствие: вы получаете семантику at-least-once — «хотя бы один раз». Дубли возможны: консьюмер сделал работу, а на ack не хватило миллисекунды до падения — сообщение уедет к соседу и будет обработано второй раз. Поэтому обработчик обязан быть идемпотентным: заведите ключ идемпотентности (message_id, идентификатор заказа) и проверяйте в базе, не делали ли это уже. «Ровно один раз» брокер вам не продаст — эту гарантию строит приложение.

Посмотреть на unacked глазами можно так:

rabbitmqctl list_queues name messages_ready messages_unacknowledged

Результат:

name    messages_ready  messages_unacknowledged
emails  1240            8

8 сообщений прямо сейчас «в руках» у консьюмеров. Это нормальное рабочее число. А вот если messages_unacknowledged растёт и не падает — консьюмеры зависли и не подтверждают. Если растёт messages_ready — просто не успеваете, нужно больше консьюмеров.

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

auto_ack=True «чтобы не разбираться». Одна строчка, из-за которой пропадают письма и платежи. Ставьте её осознанно и только там, где потеря сообщения ничего не стоит.
Ack в начале обработчика. Выглядит аккуратно, работает как auto_ack. Ack — это всегда последнее действие.
Ack внутри except с «залогируем и поедем дальше». Сообщение исчезает тихо. Иногда это осознанный выбор — но он должен быть осознанным, а не следствием привычки глушить исключения.
Ack из другого потока. pika не потокобезопасна. Если работа ушла в ThreadPoolExecutor, подтверждать надо через connection.add_callback_threadsafe(lambda: ch.basic_ack(delivery_tag=tag)). Прямой вызов из чужого потока портит состояние канала, и вы получаете загадочные обрывы соединения без внятной причины.
Долгая обработка в основном потоке. Пока обработчик считает, pika не отвечает на heartbeat, брокер считает соединение мёртвым и рвёт его — а незакрытая работа возвращается в очередь и запускается повторно. Лечится тем же выносом работы в поток.
Обработка дольше consumer_timeout. Канал закроется, сообщение уедет другому консьюмеру — и вы получите две параллельные обработки одного заказа.
requeue=True на любую ошибку. Прямая дорога в poison-loop.
Забыть про идемпотентность. At-least-once без защиты от дублей — это второе списание денег у клиента, и rollback тут делать нечем.

Итоги

auto_ack=True = at-most-once: брокер забывает сообщение в момент отправки, падение консьюмера = потеря.
Ручной ack = at-least-once: сообщение висит в unacked, пока вы не подтвердите; обрыв канала мгновенно возвращает его в очередь с флагом redelivered.
Порядок железный: получили → сделали работу → basic_ack. Подтверждать нужно тем же каналом и тем же delivery_tag.
basic_nack(requeue=True) — «попробуйте ещё», requeue=False — «сообщение битое, в dead-letter». Безусловный requeue приводит к poison-loop.
At-least-once означает дубли. Идемпотентность обработчика — обязательная часть надёжности, а не опция.
Мониторьте messages_unacknowledged: растущее и не спадающее значение — верный признак зависших консьюмеров.