RabbitMQ как инструмент «деградации с честью»

Практика outbox, гарантии доставки, DLX и схемы, которые реально спасают прод

Ситуация. REST-вызовы между сервисами начали валить друг друга: цепочки запросов, таймауты, каскадные фейлы. Мы перевели коммуникацию на RabbitMQ – и добились не «безошибочности», а предсказуемой деградации: данные не теряются, критичное – едет первым, яд быстрее восстанавливается

Ниже – только прикладное: проверенные паттерны, готовые фрагменты кода (Python / aio-pika), и схемы, которые можно сразу встраивать

База: «деградировать с честью» – это как?

Принцип простой: если брокер, потребители или сеть «сыпятся», система не ломает бизнес-операции и не теряет событий, а:

• Временно буферизует

• Расставляет приоритеты

• Отбивает яд от «ядовитых» сообщений

• Восстанавливается до согласованного состояния

В RabbitMQ это собирается из четырёх кирпичей:

1. Transactional Outbox (события в своей БД до брокера)

2. Безопасная публикация (publisher confirms + mandatory / AE)

3. Безопасное потребление (ручные ack / nack, DLX, идемпотентность)

4. Контроль нагрузки (prefetch, TTL, max-length, приоритеты / разделение очередей)

Шаг 1. Transactional Outbox – никаких потерь при падении брокера

Суть: вместе с записью бизнес-события пишем строку в outbox в той же транзакции. Фоновый воркер публикует в RabbitMQ с подтверждениями; при сбоях – ретраит. Событие не «исчезнет» между БД и брокером

DDL (PostgreSQL):

create table outbox (   id            uuid primary key,   aggregate     text not null,   event_type    text not null,   payload       jsonb not null,   created_at    timestamptz not null default now(),   status        text not null default 'NEW',      -- NEW|SENT|FAILED   attempts      int  not null default 0,   next_retry_at timestamptz );  create index on outbox (status, next_retry_at);

Воркер публикации (Python, aio-pika, подтверждения + mandatory):

import asyncio, json, uuid, asyncpg from aio_pika import connect_robust, Message, DeliveryMode, RobustConnection, RobustChannel, ExchangeType from datetime import datetime, timedelta  AMQP_URL = "amqp://user:pass@rabbit:5672/%2F" EXCHANGE = "events.topic"  async def publish_outbox_batch(pool):     conn: RobustConnection = await connect_robust(AMQP_URL)     ch: RobustChannel = await conn.channel(publisher_confirms=True)     exchange = await ch.declare_exchange(EXCHANGE, ExchangeType.TOPIC, durable=True,                                          arguments={"alternate-exchange": "events.ae"})  # AE на случай unroutable     while True:         async with pool.acquire() as db:             rows = await db.fetch("""               update outbox                  set status='IN_PROGRESS'                where id in (                  select id from outbox                   where status='NEW'                      or (status='FAILED' and (next_retry_at is null or next_retry_at <= now()))                   order by created_at                   limit 200                )              returning id, event_type, payload             """)         if not rows:             await asyncio.sleep(0.3)             continue          to_ack, to_fail = [], []         for r in rows:             body = json.dumps(r["payload"]).encode()             msg = Message(                 body,                 delivery_mode=DeliveryMode.PERSISTENT,                 message_id=str(uuid.uuid4()),                 content_type="application/json",                 headers={"event_type": r["event_type"]},             )             routing_key = f"{r['event_type']}"             try:                 # mandatory=True - UnroutableError, если некуда доставить                 await exchange.publish(msg, routing_key=routing_key, mandatory=True)                 to_ack.append(r["id"])             except Exception:                 to_fail.append(r["id"])          async with pool.acquire() as db:             if to_ack:                 await db.execute("update outbox set status='SENT' where id = any($1)", to_ack)             if to_fail:                 await db.execute("""                   update outbox                      set status='FAILED',                          attempts = attempts + 1,                          next_retry_at = now() + ((least(attempts, 10)) * interval '10 seconds')                    where id = any($1)                 """, to_fail)  async def main():     pool = await asyncpg.create_pool(dsn="postgres://user:pass@db/app")     await publish_outbox_batch(pool)  asyncio.run(main())

Что важно:

• publisher_confirms=True – ждём подтверждение брокера

• mandatory=True + alternate-exchange (AE) – гарантируем, что ни одно сообщение не уйдёт в «вакуум»

• Экспоненциальный backoff по attempts – не душим брокер при восстановлении

Шаг 2. Публикация с гарантиями: confirms, mandatory, AE

• Publisher Confirms: брокер подтверждает приём (пер-сообщение или батчем)

• mandatory: если ни одна привязка не подходит, брокер вернёт сообщение (или отправит в AE)

• Alternate Exchange (AE): запасной обменник для unroutable – складываем туда и алертим

Топология (объявление обменников и AE):

# основной обменник событий await ch.declare_exchange("events.topic", ExchangeType.TOPIC, durable=True,                           arguments={"alternate-exchange": "events.ae"}) # AE для не маршрутизованных await ch.declare_exchange("events.ae", ExchangeType.FANOUT, durable=True) # очередь-парковка для unroutable q_ae = await ch.declare_queue("events.unroutable", durable=True) await q_ae.bind("events.ae", routing_key="")

Шаг 3. Потребление без сюрпризов: ack / nack, DLX, идемпотентность

Цели:

• Не теряем сообщения

• Не зацикливаем «ядовитые»

• Обрабатываем идемпотентно (дубликаты возможны)

Очередь с DLX / TTL / ограничением:

# Dead-letter exchange await ch.declare_exchange("events.dlx", ExchangeType.TOPIC, durable=True)  args = {     "x-dead-letter-exchange": "events.dlx",     "x-dead-letter-routing-key": "dead.order",   # или по шаблону     "x-message-ttl": 10_000,                     # TTL сообщений в очереди (пример)     "x-max-length": 50_000,                      # защитимся от бесконечного роста     "x-queue-mode": "lazy",                      # крупные очереди - на диск     "x-max-priority": 10,                        # если используете приоритеты }  q = await ch.declare_queue("order.events", durable=True, arguments=args) await q.bind("events.topic", routing_key="order.*")

Идемпотентный consumer (с учётом дубликатов и «ядовитых»): как не потерять сообщение при сбое

import hashlib, aioredis  redis = await aioredis.from_url("redis://redis:6379/0")  def seen_key(message_id: str) -> str:     return f"seen:{message_id}"  async def handle(message):     # бизнес-логика     ...  async def consume(message):     async with message.process(ignore_processed=True):  # manual ack / nack внутри контекста         mid = message.message_id or hashlib.sha1(message.body).hexdigest()          # идемпотентность: уже обрабатывали?         if await redis.setnx(seen_key(mid), "1"):             await redis.expire(seen_key(mid), 24*3600)  # хранить сутки / неделю - по объёму             try:                 await handle(message)                 # ack произойдёт при выходе из контекста             except TransientError:                 # временная ошибка - вернём в очередь (с учётом TTL / DLX / повторных попыток)                 await message.nack(requeue=True)             except Exception:                 # яд - в DLX, чтобы не крутить бесконечно                 await message.reject(requeue=False)         else:             # дубликат - подтверждаем тихо             return  await q.consume(consume, no_ack=False) await ch.set_qos(prefetch_count=64)  # контроль нагрузки

Зачем DLX: «ядовитые» сообщения автоматически попадают в «кладбище» (parking lot). Там их можно руками разбирать или переигрывать через отдельный «ре-процессор»

Шаг 4. Приоритеты и разделение путей: критичное едет первым

Частая ошибка – «одна очередь на всё». В результате уведомления забивают полосу, а деньги / заказы ждут

Правильно:

• Разделяем критичное и фоновое по разным очередям / ключам

• Для фонового – TTL + max-length (готовы терять «хвост» при аварии)

• Для критичного – отдельные воркеры и меньший prefetch (быстрее реакция)

Пример:

# критичная очередь (оплата) await ch.declare_queue("billing.high", durable=True,                        arguments={"x-dead-letter-exchange": "events.dlx"}) await ch.queue_bind("billing.high", "events.topic", routing_key="billing.*")  # фоновая (email): TTL + max-length, пусть деградирует первой await ch.declare_queue("notify.low", durable=True,   arguments={     "x-dead-letter-exchange": "events.dlx",     "x-message-ttl": 60_000,     "x-max-length": 20_000 }) await ch.queue_bind("notify.low", "events.topic", routing_key="notify.*")

Набор «граблей» и как мы их обходили

• Большие очереди = антипаттерн. Делайте короткими: x-message-ttl, x-max-length, «ленивые» (x-queue-mode=lazy) только там, где осознанно готовы к диску

• Дубликаты неизбежны. Проектируйте идемпотентно (натуральные операции / UPSERT, ключи, registry processed ids)

• Один пользователь RabbitMQ для всех сервисов – плохо: лишает видимости и контроля. Делите доступы

• guest / guest в проде – никогда

• prefetch по умолчанию душит латентность: подберите set_qos экспериментально (обычно 16–256)

• Нет AE / mandatory – «чёрная дыра»

• Нет DLX – «ядовитые» крутятся бесконечно

Чек-лист внедрения

1. Включить Transactional Outbox в сервисах-продюсерах

2. Публиковать с publisher confirms + mandatory=true, настроить AE

3. Для каждой критичной линии – своя очередь, для фоновой – TTL / ограничения

4. Подписчики: ручные ack / nack, DLX, идемпотентность

5. Настроить prefetch, health-пробы, авто-reconnect

6. Метрики: ready / unacked, publish / ack rate, глубина DLQ, x-death

7. Runbook: как чистить DLQ, как «ре-проигрывать» из парковки

Итог

RabbitMQ сам по себе не «лечит» систему. Но с правильной топологией (Outbox – Confirms + AE – Idempotent Consumers + DLX – Контроль нагрузки) вы получаете контролируемую деградацию: критичные события приходят, фоновые отваливаются первыми, «ядовитые» безопасно паркуются, а данные – не теряются. Это и есть «честная» отказоустойчивость в реальной жизни