RabbitMQ

01Идея: почта для сообщений

Представьте RabbitMQ как почтовое отделение. Продюсер приносит письмо (сообщение) к стойке (exchange) с адресной наклейкой (routing key). Почта раскладывает его по почтовым ящикам (очередям) согласно правилам (bindings). Консьюмер опустошает свой ящик и расписывается за каждое письмо (acknowledgement). Если консьюмер так и не расписался, письмо возвращается на повторную доставку. Благодаря такому разделению продюсеру не нужно знать, кто обработает сообщение, сколько существует воркеров и онлайн ли они прямо сейчас: сообщения спокойно ждут в очереди.

• Сообщение: полезная нагрузка (байты, обычно JSON) плюс свойства вроде content type, message ID и заголовков.
• Очередь: упорядоченный буфер, который хранит сообщения, пока их не заберёт консьюмер.
• Connection и channel: одно TCP-соединение на процесс, а внутри него лёгкие каналы для каждого потока или задачи.

02Exchange и маршрутизация

Продюсеры никогда не публикуют напрямую в очередь — они публикуют в exchange. Тип exchange определяет маршрутизацию. Direct exchange доставляет в очереди, у которых binding key точно совпадает с routing key. Topic exchange сопоставляет шаблоны с точками: «order.*» совпадает с «order.created», а «order.#» — с любой глубиной. Fanout exchange копирует каждое сообщение во все привязанные очереди, игнорируя ключи. Headers exchange маршрутизирует по значениям заголовков. Типичная схема для событий: один topic exchange на домен (orders, users) и одна очередь на каждую задачу консьюмера (email.order-created, analytics.orders), чтобы каждый консьюмер получал свою копию и работал в своём темпе.

• Default exchange: безымянный direct exchange, который направляет сообщение в очередь с тем же именем, что и routing key.
• Немаршрутизируемые сообщения отбрасываются, если не публиковать с флагом mandatory или не настроить alternate exchange.

03Типы очередей: quorum, classic и streams

Quorum-очереди реплицируют данные на нечётное число узлов с помощью алгоритма консенсуса Raft; сообщение подтверждается только после того, как его получило большинство, поэтому оно переживает отказ узла. Это рекомендуемый выбор для всего важного. Classic-очереди живут на одном узле и не реплицируются, с тех пор как RabbitMQ 4.0 убрал classic mirrored queues; они подходят для временной или воспроизводимой работы. Streams — это append-only лог: консьюмеры читают с определённого offset, а сообщения не удаляются при потреблении, что даёт replay и очень большой fan-out, похоже на топики Kafka. Quorum-очереди также поддерживают delivery limit, который защищает от poison-сообщений, раз за разом роняющих консьюмеров.

04Надёжная публикация

Durable-очереди и persistent-сообщения переживают перезапуск брокера, но они не сообщают продюсеру, что брокер получил сообщение. Для этого используйте publisher confirms: брокер подтверждает каждое сообщение, как только оно надёжно сохранено (для quorum-очередей — реплицировано на большинство). Если подтверждение не пришло или вернулся nack, опубликуйте повторно. Давайте каждому сообщению уникальный messageId, чтобы консьюмеры могли обнаруживать дубликаты: из-за повторов доставка становится at-least-once. Объявляйте exchange и очереди в коде или в файле definitions, чтобы окружения были воспроизводимыми.

1import amqp from "amqplib";
2 
3const conn = await amqp.connect("amqp://your_user:your_password@localhost:5672");
4const ch = await conn.createConfirmChannel();
5 
6await ch.assertExchange("orders", "topic", { durable: true });
7await ch.assertQueue("email.order-created", {
8  durable: true, arguments: { "x-queue-type": "quorum" },
9});
10await ch.bindQueue("email.order-created", "orders", "order.created");
11 
12ch.publish("orders", "order.created", Buffer.from(JSON.stringify({ orderId: 42 })), {
13  persistent: true, contentType: "application/json", messageId: "order-42-created",
14});
15await ch.waitForConfirms(); // throws if the broker rejected the message

05Потребление: ack, prefetch и повторы

Консьюмеры должны использовать ручные подтверждения: ack только после того, как работа успешно выполнена. Если консьюмер упал до ack, RabbitMQ повторно доставит сообщение другому консьюмеру. Prefetch (basic.qos) ограничивает, сколько неподтверждённых сообщений держит один консьюмер; без него брокер отдаёт всё первому консьюмеру, а остальные простаивают. Начните с prefetch 10–50 и подстраивайте. При ошибке решите: вернуть в очередь при временной ошибке или отклонить без requeue, чтобы сообщение ушло в dead letter exchange (DLX) для разбора или отложенного повтора. Никогда не делайте ack до обработки и делайте обработчики идемпотентными.

1await ch.prefetch(20);
2 
3await ch.consume("email.order-created", async (msg) => {
4  if (!msg) return;
5  try {
6    const event = JSON.parse(msg.content.toString());
7    await sendOrderEmail(event.orderId); // must be idempotent
8    ch.ack(msg);
9  } catch (err) {
10    // false, false = do not requeue: route to the dead letter exchange
11    ch.nack(msg, false, false);
12  }
13}, { noAck: false });

06Dead letters, политики и эксплуатация

Dead letter exchange получает сообщения, которые были отклонены, истекли по TTL или превысили delivery limit. Настраивайте его через политику, а не жёстко прописанными аргументами очереди, потому что политики можно менять без повторного объявления очередей. Классический паттерн повторов направляет упавшие сообщения в retry-очередь с TTL, чей собственный DLX указывает обратно на рабочий exchange, — так получаются отложенные повторы. Запускайте RabbitMQ как кластер из трёх узлов для quorum-очередей, включите management-плагин для UI и HTTP API, отдавайте метрики через Prometheus-плагин, используйте TLS и отдельных пользователей для каждого приложения с ограниченными правами на vhost. Удалите стандартного пользователя guest, который и так работает только с localhost.

1# Dead-letter failed messages and cap redeliveries for all email.* queues
2rabbitmqctl set_policy email-dlx "^email\." \
3  '{"dead-letter-exchange":"dlx","delivery-limit":5}' --apply-to quorum_queues
4 
5rabbitmqctl add_vhost shop
6rabbitmqctl add_user orders_app 'use-a-strong-password'
7rabbitmqctl set_permissions -p shop orders_app '^orders.*' '^orders.*' '^(orders|email).*'

07Quorum-очереди изнутри

Quorum-очередь — это группа Raft: один лидер и фолловеры на других узлах, обычно три или пять участников. Публикации идут к лидеру, который дописывает их в свой лог и реплицирует; как только большинство записало запись на диск, лидер подтверждает её продюсеру. Если узел лидера падает, оставшееся большинство за секунды выбирает нового лидера, и консьюмеры продолжают работу. Поскольку каждое сообщение реплицируется и проходит fsync, quorum-очереди больше нагружают дисковый I/O и лучше всего подходят для очередей, которые регулярно разгребаются, а не для миллионов сообщений, лежащих днями. Они считают повторные доставки, применяют delivery limit (значение по умолчанию есть начиная с 4.0) и поддерживают приоритеты сообщений начиная с 4.0. Очень длинные очереди стоит превращать в streams.

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

08Flow control, алармы и back-pressure

RabbitMQ защищает себя сам. Когда потребление памяти на узле превышает high watermark (по умолчанию долю RAM) или свободное место на диске опускается ниже disk limit, он поднимает аларм и блокирует все публикующие соединения во всём кластере, пока ресурсы не восстановятся. Консьюмеры продолжают работать — так система разгружается. Credit-based flow control также притормаживает быстрых продюсеров, обгоняющих процессы очередей. На практике алармы — это симптом того, что очередь растёт быстрее, чем потребляется. Задавайте политики max-length или TTL для очередей, которые могут расти без ограничений, ставьте алерты на скорость роста глубины очереди, масштабируйте консьюмеров и давайте брокеру быстрые диски с запасом. Клиенты должны обрабатывать уведомление connection.blocked и останавливать или буферизовать публикацию.

09Повторы с backoff и poison-сообщения

Немедленный requeue повторяет слишком быстро и может зациклить poison-сообщение. Надёжный паттерн использует dead lettering для задержек: при ошибке консьюмер отклоняет сообщение, и оно через DLX попадает в retry-очередь с message TTL (например, 30 секунд) и без консьюмеров; когда TTL истекает, эта очередь отправляет его как dead letter обратно в основной exchange. Используйте несколько retry-очередей для экспоненциальных шагов (30 s, 5 min, 30 min) и считайте попытки в заголовке или опирайтесь на заголовок x-delivery-count в quorum-очередях. После последней попытки направляйте сообщение в parking-очередь, которую разбирает человек. Существует и плагин delayed message exchange, но он не реплицируется, поэтому для критичной работы TTL-очереди безопаснее.

1# Retry queue: holds failed messages for 30 s, then sends them back
2rabbitmqctl set_policy retry-30s "^retry\.30s$" \
3  '{"message-ttl":30000,"dead-letter-exchange":"orders"}' --apply-to queues
4 
5# Main work queues dead-letter into the retry exchange
6rabbitmqctl set_policy work-dlx "^email\." \
7  '{"dead-letter-exchange":"retry","delivery-limit":10}' --apply-to quorum_queues

10Exactly-once — миф: идемпотентность и outbox

Сеть может отказать между «обработано» и «подтверждено», а также между «опубликовано» и «confirmed», поэтому любая надёжная система работает как at-least-once. Безопасным это делают два паттерна. Transactional outbox на стороне продюсера: запишите бизнес-изменение и строку outbox в одной транзакции базы данных, а relay пусть публикует строки outbox с confirms и помечает их отправленными. Это предотвращает ситуации «заказ сохранён, а событие потеряно» и «событие отправлено, а заказ откатился». Идемпотентный консьюмер на принимающей стороне: сохраняйте ID обработанных сообщений (или используйте естественные ключи и upsert) в той же транзакции, что и побочный эффект, и пропускайте дубликаты. С обоими паттернами система ведёт себя так, будто доставка происходит ровно один раз.