Модуль: Распределённые системы · Уровень: Senior+

TL;DR#

• RabbitMQ — это smart broker / dumb consumer: брокер хранит логику маршрутизации (exchanges + bindings) и выталкивает (push) сообщения потребителям. Kafka — наоборот: dumb broker / smart consumer (pull, потребитель сам держит offset).
• Базовая модель AMQP 0-9-1: producer публикует в exchange, exchange по binding и routing key раскладывает сообщение по очередям, consumer читает из очереди. Очередь — единственное место, где сообщение реально хранится.
• Типы exchange: direct (точное совпадение routing key), topic (паттерны с * и #), fanout (broadcast, ключ игнорируется), headers (матчинг по заголовкам).
• Надёжность доставки строится на трёх китах: manual ack/nack (подтверждение от consumer), publisher confirms (подтверждение от брокера producer-у), persistent + durable + quorum queue (переживание рестарта/отказа узла).
• DLQ (dead-letter exchange) ловит сообщения, которые были отвергнуты (reject/nack без requeue), протухли по TTL, или вытеснены по max-length.
• Prefetch (basic.qos) — главный рычаг производительности: без него один медленный consumer заберёт всю очередь round-robin’ом; с prefetch=1 получается fair dispatch.
• Quorum queues (на базе Raft) — современный стандарт для надёжности. Classic mirrored queues deprecated и удалены в RabbitMQ 4.0.
• Выбор: RabbitMQ — для task/job queue, сложной маршрутизации, RPC, per-message acknowledgement. Kafka — для event streaming, replay, высокого throughput и ordered log.

Теория#

Модель AMQP 0-9-1#

AMQP 0-9-1 (Advanced Message Queuing Protocol) — это бинарный wire-протокол, реализуемый RabbitMQ по умолчанию. Не путать с AMQP 1.0 — это совершенно другой протокол (он тоже поддерживается RabbitMQ через плагин, но дефолтная модель — именно 0-9-1).

Ключевая идея, которую важно произнести на собеседовании: producer никогда не публикует напрямую в очередь. Он публикует в exchange. Exchange — это маршрутизатор без хранения; очередь — буфер с хранением.

                         binding (routing pattern)
                              |
  Producer --publish-->  [ Exchange ] --route--> [ Queue ] --deliver--> Consumer
              (routing key)      \--route--> [ Queue ] --deliver--> Consumer

Иерархия сущностей:

• Connection — TCP-соединение к брокеру (дорогое, держим долгоживущим).
• Channel — лёгкая виртуальная сессия внутри connection. Вся работа (publish, consume, declare) идёт через channel. Каналы НЕ потокобезопасны — один channel на одну горутину, не шарить между горутинами.
• Exchange — точка входа, выполняет маршрутизацию.
• Queue — FIFO-буфер (с оговорками про prefetch/requeue), хранит сообщения.
• Binding — правило «связать exchange и queue по такому-то паттерну».

Виртуальные хосты (vhost)#

vhost — это namespace/изоляция: свои exchanges, queues, permissions. Аналог отдельного логического брокера. Используется для multi-tenancy.

Exchanges: четыре типа маршрутизации#

Терминология, которую часто путают:

• routing key — атрибут сообщения, который выставляет producer при публикации.
• binding key — паттерн, указанный при создании binding между exchange и queue.

Маршрутизация = матчинг routing key (из сообщения) против binding key (из bindings).

1. Direct exchange#

Сообщение попадает в очереди, у которых binding key точно равен routing key.

binding key="error" --> queue Q1
binding key="info"  --> queue Q2
binding key="error" --> queue Q3   (одинаковый ключ -> оба получат)

publish(routing_key="error")  -> Q1 и Q3
publish(routing_key="info")   -> Q2

Дефолтный безымянный exchange ("") — это особый direct exchange: каждая очередь автоматически забиндена к нему по своему имени. Поэтому publish(exchange="", routing_key="my-queue") кладёт прямо в очередь my-queue. Это «упрощённый» режим, который и создаёт иллюзию, что публикуют «в очередь».

2. Topic exchange#

Routing key — это строка из слов, разделённых точками: logs.error.payment. Binding key — паттерн с двумя wildcard:

• * (звёздочка) — заменяет ровно одно слово.
• # (решётка) — заменяет ноль или более слов.

binding key="logs.*.payment"   matches  "logs.error.payment"      (yes)
                               matches  "logs.info.payment"       (yes)
                               NOT      "logs.error.db.payment"   (no, * = одно слово)

binding key="logs.#"           matches  "logs"                    (yes, # = ноль слов)
                               matches  "logs.error.payment.db"   (yes)

binding key="*.error.*"        matches  "app.error.db"            (yes)
                               NOT      "error.db"                (no)

binding key="#"                matches всё (поведение как fanout)

Topic — самый гибкий тип; direct и fanout логически выражаются через topic, но специализированные типы быстрее.

3. Fanout exchange#

Игнорирует routing key полностью. Копию сообщения получают все забинженные очереди. Классический pub/sub broadcast (рассылка событий, кэш-инвалидация всем нодам).

4. Headers exchange#

Маршрутизация по заголовкам сообщения вместо routing key. В binding указывается аргумент x-match:

• x-match=all — должны совпасть все указанные заголовки (AND).
• x-match=any — достаточно одного (OR).

Используется редко (медленнее topic), но полезен, когда критерий маршрутизации — несколько независимых атрибутов, плохо ложащихся в иерархический routing key.

Очереди и их свойства#

• durable — определение очереди переживает рестарт брокера (сама очередь не исчезнет). Это про метаданные очереди, НЕ про сообщения.
• persistent message (delivery_mode=2) — сообщение пишется на диск. Чтобы сообщение пережило рестарт, нужно И durable queue, И persistent message. Любое из двух по отдельности не спасает.
• exclusive — очередь видна только текущему connection и удаляется при его закрытии.
• auto-delete — удаляется, когда отвалится последний consumer.

Важно: persistent message не гарантирует «уже на диске» в момент возврата publish. Без publisher confirms сообщение может потеряться в буфере брокера до fsync.

Acknowledgements: ack / nack / reject#

Consumer должен сообщить брокеру судьбу сообщения. Это контракт at-least-once доставки.

• auto-ack (autoAck=true, в протоколе no-ack): брокер считает сообщение доставленным сразу после отправки в сокет. Если consumer упадёт, не обработав — сообщение потеряно. Быстро, но небезопасно. Использовать только когда потеря допустима (метрики, логи).
• manual ack (autoAck=false): consumer явно вызывает Ack. Пока ack не пришёл, сообщение остаётся «unacked» (в полёте). Если connection/channel закроется — брокер вернёт unacked-сообщения обратно в очередь (requeue). Это и даёт at-least-once.

Три способа подтвердить:

• basic.ack — успешно обработано, удалить.
• basic.nack — отвергнуть; поддерживает multiple и requeue. Расширение RabbitMQ (в чистом AMQP только reject).
• basic.reject — отвергнуть одно сообщение, с флагом requeue.

Флаги:

• requeue=true — вернуть сообщение в голову очереди для повторной попытки. Опасность: «poison message» зациклится навсегда (получили -> nack(requeue) -> снова получили -> …). Для отравленных сообщений requeue=false + DLQ.
• requeue=false — не возвращать; сообщение либо отправится в DLX (если настроен), либо будет отброшено.
• multiple=true — батч-ack: подтвердить это сообщение И все предыдущие неподтверждённые с меньшим delivery tag на этом канале. Сильно повышает throughput, но требует аккуратности при out-of-order обработке.

delivery tag — это монотонный per-channel идентификатор доставки (не свойство сообщения). Поэтому ack валиден только на том же канале, через который пришла доставка.

Prefetch (basic.qos): fair dispatch vs round-robin#

По умолчанию брокер раздаёт сообщения потребителям round-robin, проталкивая их сразу, не глядя на занятость consumer. Проблема: если один consumer медленный, ему всё равно набросают сообщений в локальный буфер, пока быстрые простаивают.

basic.qos(prefetch_count=N) ограничивает число неподтверждённых сообщений у consumer: брокер не пошлёт (N+1)-е, пока не получит ack за одно из N.

prefetch=1 (fair dispatch):
  Consumer A (медленный): держит 1, пока не ack-нет — больше не получает
  Consumer B (быстрый):   быстро ack-ает -> получает следующее
  => работа течёт к тому, кто свободен

• prefetch=1 — максимальная справедливость, но round-trip ack на каждое сообщение = меньше throughput.
• prefetch=больше — выше throughput за счёт пайплайнинга, но риск, что одна нода нахватает много и при падении вернёт большой requeue-батч + хуже балансировка.

QoS бывает per-channel и global (на весь канал vs на каждого consumer канала). Тюнинг prefetch — классический senior-вопрос про производительность.

Dead Letter Exchange (DLX) и причины dead-letter#

Очередь можно настроить с аргументами:

• x-dead-letter-exchange — куда отправлять «мёртвые» сообщения.
• x-dead-letter-routing-key — заменить routing key (если не задан — сохраняется исходный).

Сообщение становится dead-lettered по одной из причин (фиксируется в заголовке x-death с reason):

1. rejected — consumer сделал basic.reject или basic.nack с requeue=false.
2. expired — истёк message TTL (per-message или queue-level).
3. maxlen — очередь достигла x-max-length / x-max-length-bytes, сообщение вытеснено.
4. delivery_limit — превышен лимит доставок (актуально для quorum queues с x-delivery-limit).

Заголовок x-death накапливает историю прохождений — по нему делают retry с backoff и счётчиком попыток, отправляя в DLQ окончательно после N ретраев.

Типовой паттерн retry с задержкой: основная очередь -> DLX -> «retry»-очередь с TTL -> по истечении TTL её DLX возвращает обратно в основную. Так делают delayed retry без отдельного scheduler (либо плагин rabbitmq_delayed_message_exchange).

Message TTL#

• per-queue TTL (x-message-ttl) — TTL для всех сообщений очереди.
• per-message TTL (expiration при публикации) — для конкретного сообщения.
• При обоих действует минимальный.

Нюанс: per-queue TTL проверяется только у головы очереди — протухшее сообщение в середине будет удалено лишь когда дойдёт до головы (для классических очередей). Это влияет на момент попадания в DLX.

x-expires — TTL на саму очередь (удалить неиспользуемую очередь).

Publisher confirms#

Без подтверждений publish — fire-and-forget: producer не знает, дошло ли сообщение до очереди. Сообщение может потеряться (брокер упал до fsync, ошибка маршрутизации).

Publisher confirms — асинхронный механизм: producer переводит канал в confirm-режим (confirm.select), и брокер присылает basic.ack за каждое сообщение (по его sequence number) после того, как оно надёжно принято (для persistent — записано на диск; для зеркалируемых/quorum — реплицировано в кворуме). При проблеме приходит basic.nack.

Это даёт producer-у at-least-once на стороне публикации. Confirms поддерживают multiple (батч-подтверждение по sequence number), как и acks.

Не путать с transactions (tx.select) — AMQP-транзакции синхронны и в ~250 раз медленнее, на практике почти всегда выбирают confirms.

Mandatory flag + basic.return#

publish(mandatory=true): если сообщение не удалось замаршрутизировать ни в одну очередь (нет подходящего binding), брокер вернёт его producer-у через basic.return (с reason code, напр. NO_ROUTE). Без mandatory такое сообщение молча отбрасывается.

Это защита от «чёрной дыры»: публикуете в exchange, но binding-а нет — сообщение исчезает бесследно. Senior должен совмещать publisher confirms (дошло до брокера) + mandatory/return (нашлась ли очередь).

Был ещё флаг immediate (доставить, только если есть готовый consumer прямо сейчас) — он удалён в RabbitMQ начиная с версии 3.0. На собеседовании это хороший «детальный» факт.

Quorum queues (Raft) vs classic mirrored queues (deprecated)#

Classic mirrored queues (HA-queues, политика ha-mode) — старый способ репликации: один master + зеркала. Проблемы: split-brain, потеря сообщений при определённых сетевых разделениях, сложная синхронизация зеркал. Они deprecated с 3.x и полностью удалены в RabbitMQ 4.0.

Quorum queues — современная замена, построены на Raft (consensus-алгоритм). Каждая quorum queue — это реплицированная state machine: лидер + фолловеры, запись считается зафиксированной после подтверждения большинством (кворумом) реплик. Свойства:

• Предсказуемое поведение при сетевых разделениях (CP-выбор: при потере кворума очередь становится недоступной для записи, но не теряет/не дублирует подтверждённые данные).
• x-delivery-limit — встроенная защита от poison message (после N доставок -> dead-letter).
• Всегда durable и реплицируемы; рекомендуются для всего, где важна сохранность.
• Минусы: больше overhead памяти/диска, число реплик обычно нечётное (3/5) для кворума, не подходят для очень коротких/временных очередей и не поддерживают некоторые фичи (например, per-message priority, хотя многое уже добавлено).

Также существуют streams (RabbitMQ Streams) — append-only лог в духе Kafka, для high-throughput и replay; но это отдельная тема.

Сравнение с Kafka#

Главный нарратив для собеседования: RabbitMQ оптимизирован под «сделай эту работу один раз и убери из очереди», Kafka — под «сохрани поток событий, чтобы много разных потребителей могли читать и перечитывать его независимо».

Пример на Go (github.com/rabbitmq/amqp091-go)#

amqp091-go — официальный форк/преемник streadway/amqp. Производитель с publisher confirms и mandatory:

package main

import (
	"context"
	"log"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func publisher() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatal(err)
	}
	defer ch.Close()

	// Topic exchange, durable
	if err := ch.ExchangeDeclare(
		"logs.topic", "topic",
		true,  // durable
		false, // autoDelete
		false, // internal
		false, // noWait
		nil,
	); err != nil {
		log.Fatal(err)
	}

	// Включаем publisher confirms на канале.
	if err := ch.Confirm(false /* noWait */); err != nil {
		log.Fatal(err)
	}
	confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))
	// Возвраты по mandatory (NO_ROUTE).
	returns := ch.NotifyReturn(make(chan amqp.Return, 1))

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	err = ch.PublishWithContext(ctx,
		"logs.topic",          // exchange
		"logs.error.payment",  // routing key
		true,                  // mandatory -> вернётся через NotifyReturn, если нет binding
		false,                 // immediate (удалён, всегда false)
		amqp.Publishing{
			ContentType:  "application/json",
			DeliveryMode: amqp.Persistent, // delivery_mode=2, пишется на диск
			Body:         []byte(`{"event":"payment_failed"}`),
			Timestamp:    time.Now(),
		},
	)
	if err != nil {
		log.Fatal(err)
	}

	select {
	case c := <-confirms:
		if c.Ack {
			log.Printf("broker confirmed delivery tag=%d", c.DeliveryTag)
		} else {
			log.Printf("broker NACKed tag=%d (не сохранено)", c.DeliveryTag)
		}
	case r := <-returns:
		log.Printf("returned: %s (нет очереди для routing key)", r.ReplyText)
	case <-ctx.Done():
		log.Printf("timeout ожидания confirm")
	}
}

Потребитель с manual ack, prefetch и DLX-обработкой:

func consumer() {
	conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")
	defer conn.Close()
	ch, _ := conn.Channel()
	defer ch.Close()

	// Очередь с DLX и лимитом длины.
	args := amqp.Table{
		"x-dead-letter-exchange":    "dlx.exchange",
		"x-dead-letter-routing-key": "failed",
		"x-message-ttl":             int32(60000), // 60s TTL
		"x-max-length":              int32(10000), // вытеснение -> DLX (maxlen)
		"x-queue-type":              "quorum",     // quorum queue (Raft)
		"x-delivery-limit":          int32(5),     // poison protection
	}
	q, err := ch.QueueDeclare("payments", true, false, false, false, args)
	if err != nil {
		log.Fatal(err)
	}
	_ = ch.QueueBind(q.Name, "logs.*.payment", "logs.topic", false, nil)

	// Fair dispatch: не более 10 неподтверждённых сообщений у этого consumer.
	if err := ch.Qos(10 /* prefetchCount */, 0, false /* global */); err != nil {
		log.Fatal(err)
	}

	msgs, err := ch.Consume(
		q.Name, "worker-1",
		false, // autoAck=false -> manual ack (at-least-once)
		false, false, false, nil,
	)
	if err != nil {
		log.Fatal(err)
	}

	for d := range msgs {
		if err := process(d.Body); err != nil {
			// requeue=false -> уйдёт в DLX (reason: rejected).
			// requeue=true зациклил бы poison message.
			_ = d.Nack(false /* multiple */, false /* requeue */)
			continue
		}
		_ = d.Ack(false /* multiple */)
	}
}

func process(b []byte) error { /* ... */ return nil }

Ключевые методы для senior-словаря: Confirm, NotifyPublish, NotifyReturn, Qos, Nack(multiple, requeue), Ack(multiple), аргументы очереди x-dead-letter-exchange, x-queue-type=quorum.

Подводные камни / gotchas#

• Channel НЕ потокобезопасен. Шаринг одного канала между горутинами публикует/читает с гонками и ломает delivery tags. Один channel на горутину; connection можно шарить.
• delivery tag привязан к каналу. Ack/Nack валидны только на том канале, где пришла доставка. После реконнекта старые теги невалидны.
• durable queue ≠ persistent message. Нужны оба, иначе сообщения теряются при рестарте брокера.
• publish без confirms — fire-and-forget. Возврат из Publish не означает, что брокер сохранил/замаршрутизировал сообщение.
• mandatory забывают. Без него сообщение без binding исчезает молча. Confirms подтвердят «дошло до брокера», но не «попало в очередь».
• requeue=true + poison message = бесконечный цикл. Сообщение, которое всегда падает, будет крутиться вечно. Нужен delivery-limit (quorum) или счётчик через DLX/x-death.
• auto-ack теряет данные. Брокер ack-ает в момент отправки в сокет; падение consumer = потеря.
• prefetch по умолчанию неограничен (для старого поведения) — один consumer может нахватать всю очередь, ломая балансировку. Всегда ставьте Qos.
• Большой prefetch = большой requeue при падении и неравномерная нагрузка.
• Очередь как «общая шина» переполняется. Без max-length/TTL медленный consumer приводит к росту очереди, swap, деградации брокера (flow control / memory alarm заблокирует publishers).
• Classic mirrored queues удалены в 4.0. Код/конфиги с ha-policy сломаются — мигрируйте на quorum queues.
• Per-queue TTL чистится только с головы очереди (classic) — протухшее в середине дойдёт до DLX с задержкой.
• Memory/disk alarms блокируют publishers (TCP backpressure) — приложение «зависает» на publish, хотя это by design.

Вопросы на собеседовании#

В: В чём разница между routing key и binding key? О: Routing key — атрибут сообщения, который выставляет producer при публикации. Binding key — паттерн, заданный при создании binding между exchange и queue. Маршрутизация — это матчинг routing key против binding key по правилам типа exchange (точное равенство для direct, паттерн с */# для topic).

В: Что произойдёт с * и # в topic-биндинге logs.* для routing key logs.error.db? О: Не совпадёт. * заменяет ровно одно слово, а тут после logs. два слова. Совпало бы logs.# (# = ноль или более слов) или logs.*.*.

В: Чем publisher confirms отличаются от consumer ack? О: Это разные концы пути. Confirms — подтверждение от брокера producer-у, что сообщение надёжно принято (записано/реплицировано). Consumer ack — подтверждение от consumer брокеру, что сообщение обработано и можно удалять. Для end-to-end at-least-once нужны оба, плюс mandatory для проверки наличия очереди.

В: Зачем нужен prefetch и что даёт prefetch=1? О: Prefetch (basic.qos) ограничивает число неподтверждённых сообщений у consumer. Без него брокер раздаёт round-robin, не глядя на занятость, и медленный consumer забивается, пока быстрые простаивают. prefetch=1 даёт fair dispatch: следующее сообщение приходит только после ack предыдущего — работа течёт к свободным. Цена — больше ack round-trip и ниже throughput.

В: Назовите причины, по которым сообщение попадает в DLQ. О: (1) reject/nack с requeue=false; (2) истёк message TTL; (3) превышен max-length/max-length-bytes (вытеснение); (4) превышен delivery-limit (quorum queues). Причина пишется в заголовок x-death.

В: Как сделать retry с задержкой без внешнего планировщика? О: Через DLX + TTL: основная очередь reject-ит в retry-очередь, у которой message-TTL и DLX обратно на основную. По истечении TTL сообщение «оживает» и возвращается. Либо плагин delayed-message-exchange. Счётчик попыток ведём по x-death, после N ретраев — в постоянную DLQ.

В: Чем quorum queues лучше classic mirrored queues? О: Quorum queues на Raft: запись фиксируется кворумом реплик, предсказуемое поведение при сетевых разделениях (нет split-brain, не теряют подтверждённые данные), встроенный delivery-limit для poison messages. Mirrored queues страдали от split-brain и потери сообщений и удалены в RabbitMQ 4.0.

В: Когда выбрать Kafka вместо RabbitMQ? О: Когда нужен replay (перечитать историю по offset), очень высокий throughput, строгий ordered лог, много независимых потребителей одного потока, event sourcing/streaming/аналитика. RabbitMQ — для task queue, сложной маршрутизации, RPC, per-message ack/reject, отложенных задач, где сообщение удаляется после обработки.

В: Что делает флаг mandatory и что без него? О: При mandatory=true, если сообщение нельзя замаршрутизировать ни в одну очередь (нет binding), брокер вернёт его producer-у через basic.return (NO_ROUTE). Без mandatory такое сообщение молча отбрасывается — «чёрная дыра».

В: Сообщение помечено persistent, очередь durable. Гарантирована ли сохранность сразу после publish? О: Нет. Без publisher confirms producer не знает, что брокер успел сделать fsync. Брокер может упасть с сообщением в буфере до записи на диск. Гарантию даёт только confirm (для persistent он приходит после записи на диск / репликации кворумом).

На что копают на senior+#

• Точная семантика доставки. AMQP даёт at-least-once; exactly-once на уровне брокера нет — нужна идемпотентность consumer (dedup по message-id/business key). Умейте объяснить, где именно появляются дубликаты (requeue после падения до ack, реконнект с unacked).
• Ordering под нагрузкой. Строгий FIFO ломается при нескольких competing consumers, prefetch>1 и requeue (сообщение возвращается, но другие уже ушли вперёд). Если нужен порядок — single consumer / single active consumer / партиционирование по ключу (или Kafka).
• Backpressure и flow control. Memory/disk alarms блокируют publishers через TCP; connection.blocked. Senior понимает, что «зависший publish» — это защита брокера, и проектирует с max-length/TTL/quorum, а не безразмерные очереди.
• Connection/channel модель и реконнект. Топология (exchanges/queues/bindings) и consumers должны пересоздаваться после реконнекта; delivery tags инвалидируются. Часто используют обёртки с авто-recovery, но важно понимать, что именно теряется.
• Quorum queues внутри = Raft. Кворум, нечётное число реплик, поведение при потере большинства (становится недоступной на запись — CP). Связка с темой консенсуса.
• Сравнение архитектур. Чётко разделять smart-broker/push vs dumb-broker/pull и вытекающие последствия: replay, retention, throughput, масштабирование, ordering. Это любимый «системный» вопрос.
• Идемпотентный producer + transactional outbox. Как не потерять и не задвоить событие между БД и брокером (outbox pattern, dedup на стороне consumer) — частый follow-up.