8.3RabbitMQ и альтернативы

Сопоставь Kafka и RabbitMQ по жизненному циклу сообщения: в RabbitMQ сообщение маршрутизируется в очередь и обычно удаляется после ack, а в Kafka сохраняется в логе независимо от чтения. Разбери routing, prefetch, confirms и dead-letter retry на примерах.

Модель AMQP 0-9-1: кто здесь кто

RabbitMQ построен на открытом протоколе AMQP версии 0-9-1. Кроме передачи байтов по сети, протокол задаёт модель маршрутизации: как сообщение само находит дорогу к нужной очереди.

Эту модель проще всего понять на обычной почте. Письмо ты не несёшь адресату в квартиру, а сдаёшь в сортировочный центр, написав на конверте адрес. Дальше центр сверяется со своими таблицами и решает, в какие почтовые ящики положить копии. На языке AMQP сортировочный центр называется exchange (обменник), адрес на конверте служит routing key (ключом маршрутизации), а строчка в таблице «письма с таким адресом класть в такой-то ящик» зовётся binding (привязка). Почтовый ящик, где письмо лежит, пока его не заберут, и есть queue (очередь).

Аналогия держится почти до конца, а ломается в одном месте: сортировочный центр ничего не хранит и ничего не возвращает. Если ни одна строчка таблицы не подошла, письмо не уходит обратно отправителю и не ложится в «невостребованные», а просто исчезает.

На собесе первым делом проговори: продюсер в RabbitMQ никогда не публикует в очередь. Он публикует в exchange с каким-то routing key. Буфера в exchange нет, это чистая функция «сообщение + routing key + заголовки → список очередей». Правило, по которому очередь подписана на exchange, называется binding и само содержит binding key (или набор заголовков для headers).

Полный путь сообщения:

• Connection - одно TCP-соединение (и TLS-сессия) до брокера. Дорогое, живёт долго.
• Channel - логический подканал: внутри одного соединения их мультиплексируется много. Все операции (publish, consume, ack, qos, declare) идут через канал. Канал между горутинами не делят: в Go правило «один канал - одна горутина». amqp091-go прикрывает publish и ack мьютексом, но ответ синхронного вызова (declare, qos) на общем канале может достаться чужой горутине, а в других клиентах перемешанные фреймы рвут соединение с 505 UNEXPECTED_FRAME. Ошибка протокола закрывает канал (или всё соединение), и клиент должен уметь его пересоздать.
• Exchange маршрутизирует. Бывает direct, topic, fanout, headers; отдельно стоит системный default exchange с пустым именем "". Он работает как direct, и к нему каждая очередь автоматически привязана по своему имени. Поэтому «наивный» пример publish("", "my-queue", body) и работает без объявления обменника: ты попал в default exchange с routing key, равным имени очереди.
• Queue - единственное место, где сообщение реально лежит. Это FIFO-структура, и в classic-варианте она живёт на одном узле кластера, на том, где её объявили.
• Consumer либо подписывается (basic.consume, push-модель), либо сам опрашивает очередь (basic.get, pull). Второе почти всегда антипаттерн: round-trip на каждое сообщение.

Из этой модели следует неприятная вещь: если сообщение не совпало ни с одной привязкой, оно молча исчезает. Не в DLX и не в «unroutable», а в никуда. Чтобы такие потери увидеть, публикуют с флагом mandatory=true и слушают basic.return (в Go через ch.NotifyReturn(...)) либо вешают на exchange alternate-exchange, куда стекается вся «непонятная почта».

Механика routing key на примерах

Чаще всех работает topic: почти в любом проекте хватает одного topic-обменника на домен. Ключ состоит из слов через точку, это иерархия длиной до 255 байт. Спецсимволов всего два: * заменяет ровно одно слово, # - ноль или больше слов.

Ключ проектируют от общего к частному и кладут в него только то, на что кто-то реально захочет подписаться: домен.сущность.событие.регион. Идентификаторы (order.created.42718) класть в ключ можно, но бессмысленно: на конкретный id всё равно никто не подпишется, а таблица привязок в брокере строится как дерево и растёт.

// Declare идемпотентен, звать его можно при каждом старте.
// Повторный declare с другими параметрами вернёт PRECONDITION_FAILED и закроет канал.
err := ch.ExchangeDeclare(
    "ex.orders", // имя
    "topic",     // тип
    true,        // durable: переживёт рестарт брокера
    false,       // autoDelete: не удалять при отключении последней привязки
    false,       // internal: можно публиковать снаружи
    false,       // noWait: ждём подтверждения от брокера
    nil,
)

q, err := ch.QueueDeclare(
    "q.billing.orders", // именованная очередь: имя = контракт владельца-консьюмера
    true,               // durable
    false,              // autoDelete
    false,              // exclusive: доступна и другим соединениям
    false, nil,
)

// У очереди сколько угодно привязок, и работают они как OR, а не AND.
_ = ch.QueueBind(q.Name, "order.created.#", "ex.orders", false, nil)
_ = ch.QueueBind(q.Name, "order.paid.#",    "ex.orders", false, nil)

Подтверждения на стороне консьюмера: ack, nack, reject

Когда брокер отдаёт сообщение консьюмеру, оно не удаляется из очереди, а переходит в состояние unacked и остаётся привязанным к каналу. Каждая доставка получает delivery tag, монотонный счётчик внутри канала. Это не идентификатор сообщения и не сквозной номер: после реконнекта нумерация начинается заново. Закрыть доставку консьюмер может тремя способами:

Если канал или соединение закрывается (падение пода, сетевой обрыв, ошибка протокола), все unacked-сообщения этого канала автоматически возвращаются в очередь и будут доставлены заново с флагом redelivered=true. Так в RabbitMQ и устроен at-least-once: таймаут видимости, как в SQS, не нужен, гарантию даёт сам разрыв TCP-соединения.

Prefetch (basic.qos): сколько сообщений можно держать в руках

Prefetch («предвыборка») задаёт, сколько сообщений брокер может выдать одному консьюмеру авансом, не дожидаясь подтверждений по уже выданным. Это обратное давление из главы 8.1, сжатое в одно число: официант несёт не больше трёх тарелок, и четвёртую ему не дадут, пока он не поставит хотя бы одну.

basic.qos(prefetchSize, prefetchCount, global) задаёт максимум неподтверждённых доставок. Когда их набралось prefetchCount, этот потребитель ничего нового не получит: следующее сообщение уйдёт кому-то другому. prefetchSize (лимит в байтах) на практике не используют. global=false означает «лимит на каждого консьюмера канала», global=true - «лимит на весь канал суммарно».

По умолчанию prefetch не ограничен, и хуже дефолта не придумать: первый подключившийся консьюмер выгребает всю очередь себе в память, остальные простаивают.

Как подобрать значение? Каждое подтверждение стоит сетевого round-trip, поэтому prefetch=1 даёт идеально ровное распределение и низкий throughput: консьюмер простаивает, пока ack идёт туда, а следующее сообщение обратно. Ориентиры такие:

Requeue и бесконечный цикл переобработки

nack(requeue=true) возвращает сообщение в очередь. Вопрос в том, куда: RabbitMQ старается поставить его как можно ближе к исходной позиции, то есть практически в голову. Если консьюмер один, а ошибка постоянная (невалидный JSON, отсутствующая запись, битая схема), выходит цикл: доставка → ошибка → requeue → доставка через микросекунду. Ретраем такое не назовёшь: получается горячий цикл на 100 % CPU, который заодно держит за собой всю очередь.

// Плохо: вечный цикл на постоянной ошибке
msgs, _ := ch.Consume(q, "", false, false, false, false, nil)
for d := range msgs {
    if err := handle(d.Body); err != nil {
        d.Nack(false, true) // requeue = true
        continue            // и через мгновение снова сюда
    }
    d.Ack(false)
}

// Хорошо: разделяем транзиентные и постоянные ошибки
for d := range msgs {
    err := handle(d.Body)
    switch {
    case err == nil:
        d.Ack(false)
    case errors.Is(err, ErrPermanent), attempts(d) >= maxAttempts:
        // DLX у очереди один, и он ведёт в retry. В DLQ кладём сами.
        if publishToDLQ(ctx, ch, d) == nil {
            d.Ack(false)
        } // не вышло - не подтверждаем, сообщение вернётся
    default:
        d.Nack(false, false) // через DLX в retry-очередь с TTL, см. ниже
    }
}

Ещё одна неприятность: в AMQP нет встроенного счётчика попыток. Булево поле redelivered скажет «это не первая доставка», но номер попытки не назовёт. Считать попытки можно тремя способами:

• Через x-death: при каждом проходе через dead-lettering брокер сам наращивает count в заголовке x-death. Самый честный вариант, руками ничего писать не надо, но работает он, только если ретраи идут через DLX.
• Своим заголовком: консьюмер читает x-retry-count, увеличивает, публикует сообщение заново и подтверждает исходное. Работает всегда, но публикация + ack не атомарны: упал между ними и потерял сообщение (или продублировал, если порядок обратный).
• Внешним счётчиком в Redis по message_id. Он нужен, когда попытки считают сквозь несколько разных очередей и сервисов.

DLX - dead letter exchange

Термин dead letter пришёл с почты: так называют «мёртвое письмо», которое не удалось ни доставить, ни вернуть отправителю. Такие письма уходят в отдельный отдел, где их разбирают вручную. Брокеры взяли и слово, и идею: необработанное сообщение не выбрасывают, а откладывают в сторону, чтобы человек потом посмотрел, что с ним не так. Отсюда и DLQ (dead letter queue), «очередь битых сообщений».

DLX не «специальный тип очереди», а обычный exchange: в него очередь пересылает сообщения, которые из неё выбыли не по-хорошему. Задают его двумя аргументами при объявлении очереди, а в проде правильнее через policy, чтобы не передекларировать очередь ради смены маршрута:

args := amqp.Table{
    "x-dead-letter-exchange":    "dlx.orders",   // куда переслать
    "x-dead-letter-routing-key": "orders.dead",  // с каким ключом; без него сохранится исходный
}
ch.QueueDeclare("q.orders", true, false, false, false, args)

Причин уйти в DLX ровно четыре. На собесе назови все: кандидаты обычно помнят только первую.

При каждом прохождении брокер дописывает запись в заголовок x-death. Там лежит массив структур с полями queue, reason, count, exchange, routing-keys, time. Другого штатного счётчика попыток в AMQP нет: x-death[0].count честно говорит, сколько раз сообщение уже возвращалось через dead-lettering.

Отложенный ретрай: лестница из TTL + DLX

В RabbitMQ нет встроенной «доставки через 30 секунд». Стандартный трюк: завести очередь без консьюмеров, у которой есть x-message-ttl и DLX обратно на рабочий обменник. Сообщение полежит там положенное время, «протухнет» и вернётся в работу.

Durable, persistent, confirms: четыре независимых условия

«Сообщение переживёт рестарт брокера» требует не одной галочки, а конъюнкции четырёх условий, и каждое отвечает за свой кусок пути. Провалить достаточно одно.

Комбинации ведут себя ровно по таблице, без сюрпризов: transient-сообщение в durable-очереди пропадёт при рестарте (при нехватке памяти его могут выгрузить на диск, но это экономия памяти, а не долговечность), persistent-сообщение в non-durable-очереди пропадёт вместе с очередью.

// Полный «надёжный» publish
if err := ch.Confirm(false); err != nil { return err }          // включаем confirms на канале
confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 256)) // окно на 256 неподтверждённых
returns  := ch.NotifyReturn(make(chan amqp.Return, 8))          // сюда придут unroutable

err := ch.PublishWithContext(ctx, "x.orders", "order.created",
    true,  // mandatory: не смаршрутизировалось - верни, а не выбрось
    false, // immediate: RabbitMQ не поддерживает с 3.0, всегда false
    amqp.Publishing{
        DeliveryMode: amqp.Persistent, // = 2, иначе тело только в RAM
        MessageId:    evt.ID,          // ключ дедупликации для консьюмера
        Timestamp:    time.Now(),
        ContentType:  "application/json",
        Body:         body,
    })

c := <-confirms
if !c.Ack {
    // брокер не принял: публикуем заново, дубликаты тут норма
}

Classic, quorum, streams

• Classic queue, исходный тип, живёт целиком на одном узле. Реплицировать её умело только устаревшее зеркалирование (ha-mode): его объявили deprecated и выпилили в 4.0. Она дешёвая по ресурсам и годится для некритичных задач и очередей, которые не жалко потерять.
• Quorum queue - реплицированная очередь на Raft, современный дефолт для всего, что нельзя терять. Всегда durable, подтверждение = запись на большинство узлов, есть штатная обработка poison message через x-delivery-limit. Платить приходится диском и памятью (Raft-лог держится до snapshot), отсутствием transient-режима, exclusive-очередей и global-prefetch.
• Stream (3.9+) - append-only лог с офсетами: несколько независимых читателей, перечитывание с любой позиции, ретеншен по времени и размеру. По сути так RabbitMQ отвечает Kafka, когда нужно не «раздать задачи», а «многие читают одну ленту».

Kafka vs RabbitMQ: где проходит граница

Хуже всего ответить на собесе «Kafka быстрее». Они различаются не скоростью, а моделью хранения, и из неё выводится всё остальное. Kafka хранит упорядоченный лог, из которого чтение ничего не удаляет: позиция читателя остаётся его собственным числом. RabbitMQ держит набор очередей, из которых сообщение физически исчезает после ack, и мощный маршрутизатор перед ними.

Как формулировать выбор

• Kafka, если поток служит лентой фактов, которую читают несколько независимых систем и хотят перечитать: событийная интеграция между доменами, аналитика и стриминг, CDC из БД, аудит, материализация проекций, метрики. Признак: «нужен replay» или «потребителей будет больше, чем сейчас».
• RabbitMQ, если по потоку идут задачи: «сконвертируй видео», «отправь письмо», «сходи в 1С». Признаки: нужна сложная маршрутизация по атрибутам, per-message ретраи с задержкой, приоритеты, RPC-стиль request/reply, много мелких разнородных очередей, команда не хочет содержать Kafka.
• Оба сразу тоже нормальная архитектура, никакого хаоса в этом нет: события домена в Kafka, фоновые задачи в RabbitMQ. Ответ «взяли бы Kafka, потому что она модная» идёт в минус, ответ «зависит от того, лента это или задачи» в плюс.

NATS и JetStream

Core NATS - предельно простой брокер на Go: один статический бинарник, десятки мегабайт памяти, задержки в десятки микросекунд. Модель называется subject-based messaging: публикуют в тему вида orders.eu.created, подписываются по шаблону с * (ровно одно слово) и > (хвост целиком). И в отличие от остальных, Core NATS не хранит ничего. Если в момент публикации подписчика нет, сообщение просто исчезает: at-most-once, никаких ack. Зато из коробки есть то, чего нет у соседей: queue groups (подписчики с одинаковым именем группы делят поток, получаются competing consumers без объявления очередей) и встроенный request/reply с автоматическим inbox-subject для ответа.

JetStream добавляет поверх слой персистентности и включается флагом. С ним появляются:

• Stream - лог, захватывающий один или несколько subjects, с политикой хранения: limits (как Kafka: держим, пока влезает по max_msgs, max_bytes, max_age), interest (удаляем, когда все подписчики подтвердили) и workqueue (как очередь: сообщение удаляется, как только его подтвердил один консьюмер). Хранить можно в file или memory, реплицирует Raft (replicas: 3).
• Consumer - именованная позиция в стриме, durable или ephemeral, push или pull, с AckPolicy (none/all/explicit), AckWait, MaxDeliver, BackOff (готовая лестница задержек ретрая, которую в RabbitMQ приходится собирать из очередей) и FilterSubject. MaxAckPending работает как prefetch.
• Дедупликация из коробки: заголовок Nats-Msg-Id и duplicate_window на стриме. Брокер сам отбросит повторную публикацию с тем же идентификатором внутри окна. В Kafka ради этого включают идемпотентного продюсера, в RabbitMQ пишут руками.
• KV и Object store надстроены над стримом: key-value с watch на изменения и хранилище блобов. Часто они закрывают потребность, ради которой иначе тянули бы ещё и Redis.

rabbitmqctl set_policy dlx-all "^q\\.orders\\." \
  '{"dead-letter-exchange":"dlx.orders","message-ttl":600000,"max-length":100000}' \
  --apply-to queues

Что стоит запомнить

• RabbitMQ exchange маршрутизирует по binding и routing key в одну или несколько очередей.
• Manual ack и prefetch ограничивают незавершённую работу consumer.
• Publisher confirm говорит о подтверждении брокером публикации, а mandatory return сообщает о невозможности маршрутизации.

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

• Durable queue, persistent message и publisher confirm решают разные части сохранности.
• Nack с requeue=true при постоянной ошибке может создать горячий бесконечный цикл.
• Confirm не заменяет идемпотентность: publisher может не узнать исход после потери ответа и повторить публикацию.

if err := ch.Qos(10, 0, false); err != nil { return err }
msgs, err := ch.Consume(q, "", false, false, false, false, nil)

if err := handle(d.Body); err != nil {
    _ = d.Nack(false, true)
}

if err := handle(d.Body); err != nil {
    _ = d.Nack(false, true) // redeliver immediately
}