Не потерять сообщение в RabbitMQ: durability, confirms, ack и DLQ

В предыдущей статье серии мы подняли трёхнодовый кластер RabbitMQ с quorum-очередями: он пережил падение лидера без потери подтверждённых данных. Кластер отказоустойчивый. Но это только половина истории.

Кластер создаёт инфраструктуру. Гарантии доставки — это про то, что именно считается «подтверждённым». Сообщение может тихо потеряться ещё до того, как брокер его запишет на большинство реплик. Или после того, как consumer получит его, но упадёт до подтверждения. Обе стороны — producer и consumer — несут ответственность за целостность сообщения.

В этой статье разберём каждое звено цепочки: durability, publisher confirms, manual ack, Dead Letter Queue и паттерн отложенной доставки. Все сценарии можно воспроизвести на demo-стенде из репозитория.

В статье

• Durability и persistence: что переживает рестарт
• Гарантии producer: publisher confirms
• Транзакции vs confirms
• Гарантии consumer: manual ack и prefetch
• TTL: время жизни сообщений и очередей
• Dead Letter Queue: карантин для ядовитых сообщений
• Паттерн отложенной доставки: retry-queue
• Чеклист надёжности

Durability и persistence: что переживает рестарт

«Надёжный брокер» — это не просто «принял сообщение». Это «принял и записал так, что переживёт рестарт или краш». Для этого нужны три уровня durability — все три, потому что слабое звено рвёт цепочку.

Первый уровень: durable exchange. Exchange объявляется с флагом durable=true. Это означает, что определение exchange — его тип, привязки — сохраняется на диск и восстанавливается после рестарта ноды. Transient exchange исчезает при рестарте.

Второй уровень: durable queue. Аналогично для очереди: само определение очереди переживает рестарт. Но у quorum-очередей это свойство встроено в природу: quorum-очередь всегда durable — она хранится в Raft-журнале на диске нескольких реплик. Объявить quorum-очередь как transient невозможно.

Третий уровень: persistent message. Само тело сообщения должно быть помечено как persistent (delivery-mode=2). Иначе брокер держит его в памяти: очередь durable, а сообщения в ней — нет. После рестарта очередь будет, сообщений — не будет.

// В Go-клиенте amqp091:
amqp.Publishing{
    DeliveryMode: amqp.Persistent, // delivery-mode=2, переживает рестарт
    ContentType:  "text/plain",
    Body:         []byte(body),
}

Для quorum-очередей в RabbitMQ 4.x картина несколько отличается от classic. Quorum-очередь по природе своей реплицирует каждое сообщение в Raft-журнал большинства нод — поэтому сообщения в ней надёжны уже в силу механизма консенсуса, даже если delivery-mode не выставлен в persistent. Тем не менее delivery-mode=persistent — хорошая практика: во-первых, это явная декларация намерения; во-вторых, явная декларация типа помогает при переносе кода между кластерами с разным default_queue_type.

В итоге: durable exchange + durable queue + persistent message — три уровня, которые вместе дают «D» из ACID применительно к брокеру: однажды принятое сообщение не исчезнет при сбое.

flowchart TD
    P["Producer"]
    EX["Exchange\n(durable)"]
    Q["demo.orders\n(quorum queue,\ndurable, persistent)"]
    C["Consumer"]
    DLX["demo.dlx\n(DLX, fanout, durable)"]
    DLQ["demo.dlq\n(quorum)"]
    RQ["demo.retry\n(quorum, TTL 5s)"]

    P -->|"publish\n(persistent)"| EX
    EX -->|"routing"| Q
    Q -->|"Raft: реплики\nна большинстве нод"| Q
    Q -->|"publisher confirm ✓\n(после кворума Raft)"| P
    Q -->|"deliver"| C
    C -->|"ack ✓ — удаляем"| Q
    C -->|"nack / reject\n(requeue=false)"| DLX
    Q -->|"delivery-limit\nисчерпан"| DLX
    Q -->|"TTL истёк"| DLX
    DLX -->|"DLQ-сценарий"| DLQ
    DLX -->|"retry-сценарий"| RQ
    RQ -->|"TTL 5s истёк →\ndefault exchange"| Q

    style P fill:#f9f3e3,stroke:#8b7355
    style C fill:#f9f3e3,stroke:#8b7355
    style Q fill:#c9e4c5,stroke:#5b8a5e
    style EX fill:#c9e4c5,stroke:#5b8a5e
    style DLX fill:#e8d5c4,stroke:#a0785a
    style DLQ fill:#fce4e4,stroke:#c07070
    style RQ fill:#e4ecfc,stroke:#7090c0

Гарантии producer: publisher confirms

Без подтверждений публикация — это «выстрелил и забыл». Producer отправил сообщение, TCP-стек доставил его в сетевой буфер, брокер принял — но между «принял» и «записал на большинство реплик» может случиться failover. Producer об этом не узнает.

Publisher confirms — асинхронный протокол подтверждений на уровне AMQP-канала. После включения режима (channel.Confirm) брокер отправляет ack для каждого сообщения, которое безопасно принято: для quorum-очереди это означает запись в Raft-журнал большинства реплик. Если что-то пошло не так — брокер отвечает nack.

Producer обязан реагировать на nack и таймаут: повторить публикацию, залогировать, поднять алерт — в зависимости от критичности потока.

Три режима работы с confirms:

• Sync (синхронный) — опубликовал одно сообщение, дождался ack, опубликовал следующее. Максимальная простота. Низкий throughput: каждое сообщение = один round-trip к брокеру и ожидание записи. Именно так реализовано в demo-producer.
• Async — публикуешь поток сообщений, ack/nack приходят отдельным каналом нотификаций; producer ведёт «окно» неподтверждённых и реагирует по мере прихода ответов. Высокий throughput, сложнее логика.
• Batch — публикуешь пачку, затем ждёшь подтверждения всей пачки. Компромисс.

В demo-стенде producer по умолчанию работает с -confirms=true в sync-режиме:

В demo-стенде demo.orders — это имя и direct-exchange, и очереди; они связаны через binding с routing key demo.orders. Первый аргумент PublishWithDeferredConfirm — exchange, второй — routing key.

if *useConfirms {
    ch.Confirm(false) // включить publisher confirms на канале
}
dc, _ := ch.PublishWithDeferredConfirm("demo.orders", *queue, false, false, amqp.Publishing{
    DeliveryMode: amqp.Persistent,
    Body:         []byte(body),
})
if *useConfirms {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    ok, err := dc.WaitContext(ctx) // блокируемся до broker ack
    cancel()
    if err != nil || !ok {
        // считаем потерянным, повторяем
    }
}

Разница хорошо видна в demo: запустите producer с -confirms=false и в процессе работы убейте ноду-лидера командой bash scripts/failover.sh. Producer отчитается, что «отправил всё», но в очереди окажется меньше сообщений — тихие потери, о которых приложение не узнало. С -confirms=true confirm не пришёл → producer повторит → данные на месте.

# С подтверждениями — потерь нет даже при failover:
cd go && go run ./cmd/producer -n 200 -confirms=true

# Без подтверждений — тихие потери при убийстве ноды:
cd go && go run ./cmd/producer -n 200 -confirms=false

Транзакции vs confirms

AMQP поддерживает транзакции с 1.0: tx.select открывает транзакцию, tx.commit фиксирует её. Сообщения, отправленные внутри транзакции, либо все подтверждаются, либо откатываются.

ch.Tx()      // tx.select — перевести канал в транзакционный режим
// ... публикации ...
ch.TxCommit() // tx.commit — зафиксировать транзакцию

Надёжно? Да. Но у транзакций два серьёзных ограничения.

Синхронность. tx.commit — синхронный round-trip: producer блокируется и ждёт, пока брокер подтвердит commit. Это означает одно сообщение (или пачку) за round-trip, с полным ожиданием записи. Publisher confirms асинхронны: брокер может подтверждать сообщения пачками, пока producer продолжает публиковать. На практике confirms быстрее транзакций в разы — тот же уровень гарантии, но без полного блокирования.

Блокировка канала. Транзакционный канал нельзя использовать одновременно для нескольких команд в параллельном режиме: он блокируется на время транзакции. В Go-клиентах это означает, что транзакционный канал не может параллельно отдавать и принимать сообщения.

Когда транзакции оправданы. Если нужно атомарно опубликовать несколько сообщений в разные очереди как единый неделимый блок — именно как транзакцию. Или если confirms в конкретной ситуации неприменимы по техническим причинам. В остальных случаях publisher confirms — выбор по умолчанию.

Гарантии consumer: manual ack и prefetch

Producer подтвердил, что сообщение попало в брокер. Брокер доставил его consumer’у. Но если consumer упал после получения сообщения и до завершения его обработки — сообщение потеряно. Здесь в игру вступает acknowledgement на стороне consumer’а.

Auto-ack (autoack=true): брокер считает сообщение доставленным и удаляет его сразу при отправке consumer’у. Семантика — at-most-once: если процесс упадёт до обработки, сообщение исчезает навсегда.

Manual ack (autoack=false): сообщение остаётся в очереди в статусе «unacked», пока consumer явно не вызовет Ack. Если consumer упал без ack — брокер переотправит сообщение другому consumer’у. Семантика — at-least-once: одно сообщение может быть доставлено повторно.

At-least-once означает, что обработчик должен быть идемпотентным: повторная обработка того же сообщения не должна ломать данные. Классические приёмы — дедупликация по message-id или бизнес-ключу, INSERT ... ON CONFLICT DO NOTHING. Exactly-once в распределённой системе недостижим без высокой стоимости; практичный путь — at-least-once плюс идемпотентная обработка.

ch.Qos(prefetch, 0, false) // QoS prefetch
deliveries, _ := ch.Consume(queue, "", autoAck, false, false, false, nil)
for d := range deliveries {
    // ... обработка (идемпотентная) ...
    if !autoAck {
        d.Ack(false) // подтверждаем ТОЛЬКО после успешной обработки
    }
}

В demo: -autoack=false — manual ack, при failover сообщения «в пути» переотправятся. -autoack=true — они исчезнут.

Prefetch и справедливое распределение нагрузки

ch.Qos(prefetch, 0, false) — это AMQP basic.qos, per-consumer ограничение числа неподтверждённых сообщений, которые брокер выдаёт consumer’у одновременно.

Зачем он нужен:

• Защита от переполнения памяти. Без prefetch брокер отдаёт consumer’у всю очередь целиком. Если обработка медленная — сообщения копятся в памяти клиентского процесса.
• Справедливое распределение при нескольких consumer’ах. Без prefetch брокер отправляет сообщения по round-robin, не считаясь с тем, занят ли consumer. Быстрый consumer получит работу, медленный — тоже, и у него накопится очередь. С небольшим prefetch (например, 10) брокер выдаёт следующее сообщение только после подтверждения предыдущих — нагрузка распределяется равномернее.

Крайности одинаково плохи:

• prefetch=1 — максимальная справедливость, но каждое сообщение требует отдельного round-trip на ack; throughput падает.
• prefetch=0 (без ограничения) или очень большое значение — один «жадный» consumer захватит всю очередь, не давая другим consumer’ам получать сообщения. Если этот consumer медленный или упал — все его unacked-сообщения вернутся в очередь разом, и другие получат их только тогда.

Типичное значение в production — десятки. В demo-стенде по умолчанию -prefetch=10. Правильное значение зависит от времени обработки одного сообщения и числа consumer’ов.

TTL: время жизни сообщений и очередей

RabbitMQ позволяет задавать время жизни как для отдельных сообщений, так и для очередей целиком.

Message TTL (x-message-ttl) — задаётся на уровне очереди в аргументах. Сообщения, которые провели в очереди дольше указанного числа миллисекунд, считаются «протухшими». Если у очереди настроен Dead Letter Exchange (DLX), протухшее сообщение не удаляется, а направляется туда.

// Аргумент при объявлении очереди
{ "x-message-ttl": 5000 }  // 5 секунд

Можно задать TTL и на уровне отдельного сообщения — в expiration при публикации (строковое значение в миллисекундах). Если задан и очередной, и per-message TTL, применяется меньший из двух.

Queue TTL (x-expires) — если к очереди не было обращений (чтений или публикаций) дольше указанного времени, очередь автоматически удаляется. Полезно для временных очередей в RPC-паттернах или при создании очередей на лету: они не накапливаются и не засоряют брокер.

Типовые применения message TTL:

• Retry-queue — временная пауза перед повторной попыткой (детально рассмотрим в следующем разделе).
• Ограничение актуальности данных — котировки, нотификации, данные геолокации: если сообщение не забрали за N секунд, оно уже неактуально.
• Защита от переполнения — в сочетании с x-max-length: старые сообщения вытесняются, очередь не растёт бесконечно.

Dead Letter Queue: карантин для ядовитых сообщений

Dead Letter Exchange (DLX) — это exchange, куда брокер автоматически перенаправляет сообщения, когда с ними что-то пошло не так. Три основные причины попадания в DLX:

1. Consumer сделал nack или reject с requeue=false.
2. Сообщение протухло по TTL.
3. У quorum-очереди превышен delivery-limit — сообщение доставлялось и отклонялось слишком много раз.

При явном nack/reject с requeue=false сообщение уходит в DLX немедленно, минуя счётчик x-delivery-count; delivery-limit тут не задействован. delivery-limit — это защита от бесконечного цикла redelivery при закрытии канала/соединения без ack.

Третий случай — это защита от «ядовитых сообщений» (poison messages). Если consumer падает на конкретном сообщении — соединение рвётся с unacked-сообщением, либо consumer отклоняет его через basic.reject — сообщение снова возвращается в очередь, блокируя обработку других. delivery-limit — предохранитель: после N таких доставок сообщение уходит в DLX автоматически. Важно: явный basic.nack с requeue=true не увеличивает счётчик x-delivery-count и не продвигает к лимиту — quorum-очередь трактует его как «отложи, но не считай неудачей».

В demo-стенде политика dlx-orders (cluster/definitions.json) настроена так:

{
  "name": "dlx-orders",
  "pattern": "^demo\\.orders$",
  "definition": {
    "dead-letter-exchange": "demo.dlx",
    "delivery-limit": 5
  }
}

delivery-limit=5 означает 5 повторных доставок (redeliveries) плюс первичная доставка — итого 6 попыток обработки. После этого сообщение уходит в demo.dlx (fanout, durable).

Что увеличивает x-delivery-count в quorum-очереди (RabbitMQ 4.3). Счётчик растёт при: basic.reject с любым requeue, закрытии канала или соединения с unacked-сообщением, AMQP 1.0 modified(delivery_failed=true). Счётчик не растёт при: basic.nack с requeue=true, AMQP 1.0 released, AMQP 1.0 modified(delivery_failed=false). Документация формулирует так: «Only actual redeliveries (such as via reject, AMQP 1.0 modify with delivery_failed=true, or channel/session termination with pending unacknowledged messages) increment the delivery count.»

Demo-флаг -nack на деле симулирует путь, растящий счётчик: закрытие соединения с unacked-сообщением. Это наглядно демонстрирует механизм poison-message protection через delivery-limit.

Как настроить DLQ:

# Скрипт dlq-demo.sh (используется rabbitmqadmin v2):
bash scripts/dlq-demo.sh

Скрипт создаёт очередь demo.dlq (quorum, durable) и привязывает её к exchange demo.dlx.

# Шаг 1: опубликовать несколько сообщений
cd go && go run ./cmd/producer -n 3

# Шаг 2: запустить consumer в режиме симуляции падения (~15 секунд)
# В логах: каждое сообщение доставляется повторно (#1, #2, #3, #4, #5, #6)
go run ./cmd/consumer -nack

# Шаг 3: убедиться, что сообщения появились в demo.dlq
docker exec rabbit1 rabbitmqctl list_queues name messages | grep demo

После 5 redeliveries каждое сообщение автоматически уходит в demo.dlx → demo.dlq. В Management UI видно, как очередь demo.orders опустошается, а в demo.dlq накапливаются сообщения.

Паттерн отложенной доставки: retry-queue

DLQ — это карантин для окончательно «плохих» сообщений. Но часто сбой обработки временный: база данных недоступна секунду, внешний API лежит минуту. В таких случаях нужно не убрать сообщение в DLQ навсегда, а дать ему «отдохнуть» и попробовать снова. Это паттерн retry-queue с отложенной доставкой.

Схема из demo-стенда:

demo.orders → (nack, requeue=false) → demo.dlx → demo.retry (TTL 5s) → demo.orders

Механизм работает так:

1. Consumer делает nack без requeue — симулирует временный сбой.
2. Брокер пересылает сообщение в demo.dlx (dead-letter-exchange из политики dlx-orders).
3. demo.dlx — fanout, его подписчик — demo.retry (quorum, x-message-ttl=5000).
4. Через 5 секунд TTL истекает. demo.retry сконфигурирована с x-dead-letter-exchange="" и x-dead-letter-routing-key=demo.orders — это означает dead-letter в default exchange с routing key demo.orders.
5. Сообщение снова появляется в demo.orders — готово к следующей попытке.

Конфигурация demo.retry из definitions.json:

{
  "name": "demo.retry",
  "arguments": {
    "x-queue-type": "quorum",
    "x-message-ttl": 5000,
    "x-dead-letter-exchange": "",
    "x-dead-letter-routing-key": "demo.orders"
  }
}

# Запустить демонстрацию retry-сценария:
bash scripts/retry-demo.sh

Native delayed retry в RabbitMQ 4.3 (только quorum-очереди)

RabbitMQ 4.3 добавил встроенный механизм отложенных повторов для quorum-очередей — без плагинов и без ручного создания retry-очередей. Это самый простой путь, если вы работаете с quorum-очередями на версии 4.3+.

Механизм управляется аргументами очереди или policy-ключами:

Формула задержки: delay = min(min_delay × delivery_count, max_delay). Задержка линейно растёт с числом попыток: при min=2000 и max=10000 получается 2 с → 4 с → 6 с → 8 с → 10 с → 10 с…

Типы повтора:

• all — повторять все возвращённые сообщения.
• failed — только те, где x-delivery-count растёт: basic.reject, закрытие соединения с unacked.
• returned — только те, где счётчик не растёт: basic.nack с requeue=true, AMQP 1.0 released.

Пример через policy:

rabbitmqctl set_policy native-retry "^demo\.orders$" \
  '{"delayed-retry-type":"failed","delayed-retry-min":2000,"delayed-retry-max":10000}' \
  --apply-to queues

Или аргументами при объявлении очереди:

{
  "x-queue-type": "quorum",
  "x-delayed-retry-type": "failed",
  "x-delayed-retry-min": 2000,
  "x-delayed-retry-max": 10000
}

Demo-скрипт scripts/delayed-retry-demo.sh демонстрирует работу этого механизма: consumer рвёт соединение с unacked-сообщением (тип политики all; путь, растящий x-delivery-count — закрытие соединения/reject), задержка перед повтором реально растёт: 2 с → 4 с → 6 с.

Что выбрать на практике:

• RabbitMQ 4.3, quorum-очереди — native delayed retry проще и не требует дополнительных очередей.
• Переносимость или classic-очереди — TTL retry-queue: работает на любой версии RabbitMQ и любом типе очереди (подробнее — ниже).
• Динамическая задержка вне quorum — плагин rabbitmq-delayed-message-exchange.

Экспоненциальный backoff несколькими retry-очередями

Одна retry-очередь с фиксированным TTL — это простейший вариант. В production часто нужен экспоненциальный backoff: первая попытка через 5 секунд, вторая через 30, третья через 5 минут. Это реализуется несколькими retry-очередями с разными TTL:

demo.retry.5s   (x-message-ttl=5000)   → demo.orders
demo.retry.30s  (x-message-ttl=30000)  → demo.orders
demo.retry.5m   (x-message-ttl=300000) → demo.orders

Направление в нужную retry-очередь реализуется двумя способами. Первый — consumer сам republish’ит сообщение: вместо nack он явно публикует его в нужный exchange с нужным routing key, выбирая retry-очередь на основе числа предыдущих попыток из заголовка x-death. Второй — маршрутизация через DLX: при dead-lettering брокер переотправляет сообщение в DLX с его исходным routing key (или с dead-letter-routing-key, если он задан на очереди), а bindings DLX направляют его в соответствующую retry-очередь. Важно: у AMQP basic.nack нет параметра routing key — его параметры ограничены delivery-tag, multiple и requeue. Любая маршрутизация при dead-lettering определяется конфигурацией очереди и exchange, а не содержимым nack. Это более сложная логика, но результат — сообщение не блокирует очередь и не создаёт лишнюю нагрузку при частых ретраях.

Delayed Message Exchange Plugin

Альтернатива ручным retry-очередям — плагин rabbitmq-delayed-message-exchange. Он добавляет тип exchange x-delayed-message, который задерживает маршрутизацию сообщения на указанное число миллисекунд. Consumer публикует сообщение обратно в этот exchange с заголовком x-delay: 5000, и брокер сам задерживает доставку.

Преимущества: не нужно создавать несколько retry-очередей, задержка задаётся динамически для каждого сообщения. Недостатки: плагин не входит в стандартную поставку RabbitMQ, задержки хранятся в памяти ноды-лидера (риск потерь при failover), и он менее прозрачен с точки зрения наблюдаемости. На RabbitMQ 4.3 с quorum-очередями нативный delayed retry проще и надёжнее; ручные TTL retry-очереди ценны своей переносимостью (любая версия, любой тип очереди); плагин — вариант для случаев вне quorum, где нужна динамическая задержка.

Нюансы и предостережения

Петля без DLQ. В demo-стенде demo.dlx — fanout. При retry-сценарии его единственный подписчик — demo.retry. Если delivery-limit=5 исчерпан (после многократных циклов retry), сообщение после истечения TTL вернётся в demo.orders, снова будет отклонено и снова попадёт в demo.retry — потенциальная бесконечная петля. В production к demo.dlx привязывают оба подписчика: demo.retry для временных ошибок и demo.dlq для окончательно «убитых» сообщений — тех, что снова достигнут delivery-limit и снова уйдут в DLX. Как именно разделить — через разные routing key (если DLX не fanout) или логику в consumer’е.

Изоляция сценариев. Скрипты dlq-demo.sh и retry-demo.sh конкурируют за подписчиков demo.dlx. Запускайте их раздельно, выполняя между прогонами docker compose down -v для сброса привязок и состояния очередей.

Чеклист надёжности

Итог в виде конкретных пунктов — всё, что должно быть выполнено, чтобы сообщение не терялось.

Durability и хранение:

• Durable exchange + durable queue + delivery-mode=persistent.
• Для quorum-очередей: x-queue-type=quorum или vhost default_queue_type=quorum. Для RabbitMQ 4.x — только так, не через policy-ключ queue-type.
• Три ноды минимум. Защита от split-brain — автоматически через Raft-кворум (cluster_partition_handling в 4.3 удалён, не нужен).

Producer:

• Publisher confirms включены. При nack или таймауте — повтор с логированием.
• Для нагруженных систем — async или batch confirms вместо sync.
• Клиент знает все узлы кластера (флаг -urls) или за кластером стоит балансировщик.

Consumer:

• Manual ack (autoack=false). Ack вызывается только после успешной обработки.
• Обработчик идемпотентен: повторная доставка не нарушает данные.
• Разумный prefetch — десятки. Не 1 (низкий throughput) и не 0 (жадный consumer).

Устойчивость к ошибкам:

• DLX настроен на основных очередях (dead-letter-exchange).
• delivery-limit задан для quorum-очередей: защита от poison-message.
• DLQ существует и привязана к DLX: сообщения не теряются при исчерпании лимита.
• Для временных сбоев — native delayed retry (RabbitMQ 4.3+, quorum), retry-queue с TTL (переносимый паттерн) или delayed-message-exchange plugin.
• Message TTL (x-message-ttl) задан там, где устаревшие данные не нужны.

Мониторинг:

• Алерт на рост глубины DLQ: сигнал, что producer или consumer проблемный.
• Алерт на unacked/lag: consumer не подтверждает или завис.
• Алерт на потери подтверждений у producer’а.

В следующей статье серии разберём межкластерное взаимодействие: Federation и Shovel — как связать независимые кластеры в разных регионах и чем эти подходы различаются. Следите за «RabbitMQ: Federation и Shovel».

Документация и первоисточники

• RabbitMQ Publisher Confirms
• Consumer Acknowledgements and Prefetch
• Consumer Prefetch (QoS)
• Dead Letter Exchanges
• Quorum Queues
• Quorum Queues: Delayed Delivery (4.3+)
• TTL
• Production Checklist
• Demo-стенд ha-cluster