Модуль 1. RabbitMQ — классика с характером

Цель модуля: довести RabbitMQ от «hello world» до продакшена: свободно ориентироваться в AMQP-модели (connection → channel → exchange → queue), выбирать тип exchange осознанно, строить надёжную доставку (publisher confirms, ack/nack/reject, prefetch), использовать DLX/TTL/priority как конструктор, понимать, что происходит в кластере, и собрать лабораторную «Надёжный воркер».

Перед началом: Модуль 1 курса «Фундамент» (гарантии доставки, идемпотентность), Docker. Инструменты: docker-compose (rabbitmq:3.13-management), github.com/rabbitmq/amqp091-go, Redis (для лабораторной). Трудоёмкость: ~12–15 ч (теория 5 ч · практика 7 ч · квиз 1 ч).

📖 1.1. AMQP-модель: connection, channel, exchange, queue, binding

Карта протокола

AMQP 0.9.1 — протокол, на котором говорит RabbitMQ. Пять сущностей, которые нужно держать в голове всегда:

text
[Publisher] --TCP--> [Connection] --logical--> [Channel]
                                                   |
                                                   v
                                             [Exchange] --binding (routing key)--> [Queue] --push--> [Consumer]

Ключевая идея: продюсер никогда не публикует в очередь напрямую. Он публикует в exchange с routing key, а binding'и решают, куда сообщение попадёт. Это развязывает продюсера от консьюмеров: можно добавить новую очередь-подписчика, не трогая код продюсера.

Почему семантики отличаются от NATS и Kafka (важно для всего курса)

RabbitMQ — наш первый брокер, и в нём проще всего перепутать ментальные модели. Зафиксируем таблицу, к которой будем возвращаться в курсах трека:

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

Минимальный продюсер на amqp091-go

go
package main

import (
	"context"
	"log"

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

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

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

	// durable=true: exchange и очередь переживут рестарт брокера
	if err := ch.ExchangeDeclare("orders", "topic", true, false, false, false, nil); err != nil {
		log.Fatal("declare exchange:", err)
	}
	q, err := ch.QueueDeclare("orders.payment", true, false, false, false, nil)
	if err != nil {
		log.Fatal("declare queue:", err)
	}
	if err := ch.QueueBind(q.Name, "order.created.*", "orders", false, nil); err != nil {
		log.Fatal("bind:", err)
	}

	err = ch.PublishWithContext(context.Background(),
		"orders", "order.created.eu", // exchange, routing key
		false, false,                 // mandatory, immediate
		amqp.Publishing{
			ContentType:  "application/json",
			DeliveryMode: amqp.Persistent, // писать на диск
			Body:         []byte(`{"order_id":"o-1","amount":100}`),
		})
	if err != nil {
		log.Fatal("publish:", err)
	}
}

Обратите внимание на DeliveryMode: amqp.Persistent — он работает только в паре с durable очередью. Сообщение «персистентное» в transient-очереди при рестарте брокера всё равно исчезнет. Это пара, а не два независимых флага — классическая находка на собеседовании.

💻 Правило пула connection/channel (оптимизация №1)

Две ошибки, которые мы видим в каждом втором код-ревью:

1. Connection на каждый запрос. AMQP-handshake дорогой (несколько RTT), TLS — ещё дороже. Правильно: один connection на процесс (плюс запасной для админ-задач) с автопереподключением.
2. Один канал на всё. Channel не потокобезопасен: публикация из нескольких горутин в один канал даёт interleaving фреймов и панику в библиотеке. Правильно: канал на горутину-издателя или мьютекс/очередь сериализации, а для консьюмеров — по каналу на каждого.

go
type AMQPPool struct {
	conn   *amqp.Connection
	mu     sync.Mutex
	conns  []*amqp.Connection // для тяжёлых воркеров можно держать 2-3
}

// PublishOne даёт канал под одну горутину-издателя.
// Вызывающий обязан закрыть канал: ch.Close().
func (p *AMQPPool) Channel() (*amqp.Channel, error) {
	return p.conn.Channel()
}

Плюс настройки, которые стоит выставить всегда: heartbeat 10–30 секунд (amqp.DialConfig), лимит каналов на connection (по умолчанию в RabbitMQ 2047 на подключение) и reconnect с экспоненциальным backoff на уровне приложения — amqp091-go не переподключается сам.

📖 1.2. Четыре типа exchange и лабораторки к ним

Лабораторка A: direct

Продюсер публикует логи с severity в ключе. Очередь logs.error биндится на error — получает только ошибки.

go
_ = ch.ExchangeDeclare("logs", "direct", true, false, false, false, nil)
q, _ := ch.QueueDeclare("logs.error", true, false, false, false, nil)
_ = ch.QueueBind(q.Name, "error", "logs", false, nil)

_ = ch.PublishWithContext(ctx, "logs", "error", false, false,
	amqp.Publishing{Body: []byte("disk full")})

Лабораторка B: fanout

Каждый сервис — своя очередь, привязанная к одному fanout-exchange. Сообщение копируется во все очереди. Удалил свою очередь — не затронул остальных. Проверьте: подписчики, подключённые после публикации, сообщение не получат — fanout не хранит историю. Это pub/sub «в моменте».

Лабораторка C: topic

Паттерны binding'ов для exchange orders и ключей вида order.created.eu:

* — ровно один сегмент, # — ноль и больше. Точка — разделитель сегментов. Проверьте все четыре биндинга одним скриптом: это 20 строк кода и самый частый вопрос на интервью по RabbitMQ.

Лабораторка D: headers

go
_ = ch.ExchangeDeclare("docs", "headers", true, false, false, false, nil)
q, _ := ch.QueueDeclare("pdf.invoices", true, false, false, false, nil)
_ = ch.QueueBind(q.Name, "", "docs", false, amqp.Table{
	"x-match": "all", // должны совпасть все заголовки биндинга
	"format":  "pdf",
	"type":    "invoice",
})

_ = ch.PublishWithContext(ctx, "docs", "", false, false, amqp.Publishing{
	Headers: amqp.Table{"format": "pdf", "type": "invoice"},
	Body:    body,
})

Headers-exchange редкий гость в проде: он медленнее topic (сравнение таблиц заголовков), а почти всё его умеет topic + поля в payload. Держите его в уме для случаев, когда атрибутов маршрутизации много и они независимы.

📖 1.3. Надёжность: publisher confirms, consumer ack/nack/reject, prefetch

Publisher confirms: брокер подтвердил — значит записал

Publish в amqp091-go возвращает nil, когда фрейм ушёл в TCP-буфер, а не когда брокер его принял. Чтобы знать, что сообщение дошло и записано, включаем publisher confirms:

go
type ReliablePublisher struct {
	ch       *amqp.Channel
	confirms chan amqp.Confirmation
}

func NewReliablePublisher(conn *amqp.Connection) (*ReliablePublisher, error) {
	ch, err := conn.Channel()
	if err != nil {
		return nil, err
	}
	p := &ReliablePublisher{
		ch:       ch,
		confirms: ch.NotifyPublish(make(chan amqp.Confirmation, 256)),
	}
	if err := ch.Confirm(false); err != nil { // переводим канал в confirm-режим
		return nil, err
	}
	go func() {
		for c := range p.confirms {
			if !c.Ack {
				log.Printf("брокер НЕ принял сообщение: delivery-tag=%d", c.DeliveryTag)
			}
		}
	}()
	return p, nil
}

func (p *ReliablePublisher) Publish(ctx context.Context, exchange, key string, body []byte) error {
	dfc, err := p.ch.PublishWithDeferredConfirmWithContext(
		ctx, exchange, key, true /* mandatory */, false,
		amqp.Publishing{
			ContentType:  "application/json",
			DeliveryMode: amqp.Persistent,
			Body:         body,
		})
	if err != nil {
		return err
	}
	wctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	return dfc.WaitContext(wctx) // ждём confirm с таймаутом
}

Два тонких момента:

• mandatory=true. Если сообщение не сматчилось ни с одним binding, брокер вернёт его продюсеру через basic.return (читайте ch.NotifyReturn). Без mandatory unroutable-сообщение молча исчезает — это баг №1 в клиентских багрепортах.
• Confirm «на каждое» vs «на батч». WaitContext на каждое сообщение режет throughput. Для потоков, где потеря отдельных сообщений недопустима, но скорость важна, ждите confirms батчами: отправили N, получили N confirmation-фреймов. c.DeliveryTag растёт монотонно — по нему видно, какие сообщения в батче подтверждены.

Consumer ack / nack / reject

go
if err := handle(ctx, d.Body); err != nil {
	if isPermanent(err) {
		_ = d.Nack(false, false) // в DLX — это финал для сообщения
	} else {
		_ = d.Nack(false, true)  // повторим... и здесь начинается зло
	}
	return
}
_ = d.Ack(false)

⚠️ Ловушка «nack с requeue=true»: сообщение, падающее каждый раз, возвращается в голову очереди и доставляется снова. Воркер крутится в бесконечном цикле, очередь не двигается, консьюмер сгорает. Лечение — только через подсчёт попыток (x-death header) или DLX (см. 2.4).

Prefetch (basic.qos): ручка потоковости

autoAck=false без prefetch — катастрофа: брокер шлёт всё подряд в память консьюмера. Prefetch ограничивает число неакnowledged сообщений на канал:

go
if err := ch.Qos(10, 0, false); err != nil {
	return err // prefetch=10 сообщений на канал
}

Тюнинг prefetch (оптимизация №2): прикидывайте по формуле

text
prefetch ≈ (время обработки, сек) × (сообщений/сек, целевой темп) × 1.5

• prefetch = 1 — честный round-robin, но RTT на каждое сообщение убивает throughput. Только для равномерных быстрых задач.
• 10–30 — типичный продакшен для задач с обработкой 50–500 мс.
• > 100 — почти безлимит: один медленный воркер забирает пачку, а соседние простаивают (и растёт риск потерять много in-flight при падении).

Замеряйте на своей нагрузке: prefetch — это баланс между round-robin-честностью и пропускной способностью. Заодно он и есть backpressure из Модуля 1: воркер не возьмёт больше, чем может.

📖 1.4. DLX, TTL, приоритеты и отложенные задачи

Dead Letter Exchange: куда попадают «неудобные» сообщения

Очередь может объявить DLX — exchange, куда улетают сообщения, которые:

• отклонены (nack/reject с requeue=false);
• просрочились по TTL;
• не поместились в очередь (переполнение по x-max-length).

go
// Главная очередь: всё непереработанное уходит в retry.dlx
_, err = ch.QueueDeclare("tasks", true, false, false, false, amqp.Table{
	"x-dead-letter-exchange":    "retry.dlx",
	"x-dead-letter-routing-key": "retry",
})

В DLX-очереди сообщение получит заголовок x-death (причина, счётчик, время) и x-first-death-reason — по ним воркер отличает первую попытку от пятой.

TTL и «отложенные задачи» на чистом RabbitMQ

Тот же механизм — готовый конструктор отложенных задач:

text
tasks.publish.later (ttl=5m, dlx=tasks.direct) --dead-letter--> tasks (воркер)

Задачу кладём в очередь с x-message-ttl, она лежит пять минут, потом dead-letter'ится обратно в рабочую очередь. Никаких кронов и внешних шедулеров.

⚠️ Ловушка per-message TTL: просрочка проверяется с головы очереди. Если в голове сообщение с TTL 10 минут, то застрявшие за ним сообщения с TTL 30 секунд не уйдут в DLX раньше, несмотря на истёкший TTL. Поэтому для отложенных задач либо очередь на каждый интервал задержки (delay.30s, delay.5m), либо плагин rabbitmq_delayed_message_exchange (он хранит таймеры не в очереди, но при миллионах таймеров становится узким местом и единой точкой отказа).

Priority queues

go
_, err = ch.QueueDeclare("support.tickets", true, false, false, false, amqp.Table{
	"x-max-priority": int8(5),
})
// публикация с приоритетом:
_ = ch.PublishWithContext(ctx, "", "support.tickets", false, false, amqp.Publishing{
	Priority: 5,
	Body:     body,
})

x-max-priority — максимум, выше 5–10 реальные приоритеты сливаются. Помните про голодание: если поток сообщений с приоритетом 9 непрерывен, сообщения с приоритетом 1 могут не доставляться неопределённо долго. Приоритетные очереди — не лекарство от плохого дизайна SLA, а способ заложить его в очередь.

📖 1.5. Кластеризация и quorum queues

Что такое кластер RabbitMQ

Несколько узлов с общими метаданными (обменники, биндинги, политики, юзеры): очередь объявлена на одном узле — видна со всех. По умолчанию содержимое очереди живёт на одном узле (там, где её объявили мастер). Кластер даёт отказоустойчивость метаданных, но не данных очереди.

yaml
# docker-compose.yml: 3-узловой кластер
x-rabbit-base: &rabbit-base
  image: rabbitmq:3.13-management
  environment:
    RABBITMQ_ERLANG_COOKIE: "course-mq-cookie" # только для песочницы

services:
  rabbit1:
    <<: *rabbit-base
    hostname: rabbit1
    ports: ["5672:5672", "15672:15672"]
  rabbit2:
    <<: *rabbit-base
    hostname: rabbit2
    depends_on: [rabbit1]
  rabbit3:
    <<: *rabbit-base
    hostname: rabbit3
    depends_on: [rabbit1]

bash
# на rabbit2 и rabbit3 (запустить после rabbit1):
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit1@rabbit1
rabbitmqctl start_app

Quorum queues: консенсус вместо «магии зеркал»

Раньше отказоустойчивость очередей делали зеркалированием (classic mirrored queues) — сегодня это deprecated. Стандарт — quorum queues: очередь-группа на N узлах с Raft-репликацией. Запись подтверждается, когда большинство узлов группы записало сообщение.

go
q, err := ch.QueueDeclare("orders.quorum", true, false, false, false, amqp.Table{
	"x-queue-type": "quorum",
})

Что важно понимать глубже, чем «включил и работает»:

1. Кворум = N/2 + 1. Группа из 3 узлов переживает падение одного. Из 5 — двух. При потере кворума очередь недоступна целиком (на запись и чтение) — это цена консистентности, и её надо закладывать в алерты.
2. Группа не обязательно на всех узлах. x-quorum-group-size (или политика) позволяет держать группу из 3 на кластере из 10 — иначе каждое подтверждение ждёт 10 узлов и весь кластер деградирует.
3. Publisher confirms с quorum-очередью означают ровно то, что написано: сообщение реплицировано на большинство. Это самый близкий RabbitMQ аналог acks=all из Kafka (курс Kafka) — не случайно семантики совпадают.
4. Quorum-очередь требует больше ресурсов (диск, CPU на Raft) и не поддерживает some classic-фичи (например, x-max-priority). Для короткоживущих scratch-очередей classic по-прежнему норм.

Оптимизация №3 — lazy-режим для больших бэклогов. Классическая очередь держит сообщения в памяти, сбрасывая на диск под давлением. С 3.12 поведение примерно «lazy по умолчанию», но для очередей с гигабайтными бэклогами полезно помнить про x-queue-mode: lazy: минимум RAM, чуть выше латентность. Если очередь регулярно раздувается — сначала спросите себя, почему консьюмеры не успевают (курс OrderFlow целиком про это), потом настраивайте режим.

📖 1.6. Лабораторная: «Надёжный воркер»

Собираем всё модуля в один сервис, который переживёт краш, повторные доставки и падающий upstream. Требования:

1. Публикация с confirms, сообщения персистентные, очередь durable.
2. Воркер: manual ack, конечный prefetch, graceful shutdown (доставить in-flight).
3. Ошибки: 2 повторные попытки с экспоненциальным backoff (через DLX-таймеры), потом финальная DLQ.
4. Идемпотентность: повторная обработка не должна менять результат (Redis).

Топология

text
nack(requeue=false)
 tasks.main ────────────────────────────────► retry.dlx (topic)
     ▲                                          │ key=retry.1s │ key=retry.5s
     │ dead-letter (обратно в tasks.main)       ▼              ▼
     └──────────────────────────── retry.1s (ttl=1s)  retry.5s (ttl=5s)
                                                  │              │
                                                  └──────┬───────┘
                                              x-death >= 2 (маршрутизация по счётчику)
                                                         ▼
                                                  tasks.dlq (разбор руками)

Воркер с graceful shutdown и backoff

go
package worker

import (
	"context"
	"errors"
	"log"
	"time"

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

// Run потребляет задачи, пока не придёт сигнал завершения.
// In-flight сообщение дорабатывается (или nack'ается) до выхода.
func Run(ctx context.Context, conn *amqp.Connection) error {
	ch, err := conn.Channel()
	if err != nil {
		return err
	}
	defer ch.Close()

	if err := ch.Qos(10, 0, false); err != nil {
		return err
	}
	msgs, err := ch.Consume("tasks.main", "reliable-worker", false, false, false, false, nil)
	if err != nil {
		return err
	}

loop:
	for {
		select {
		case <-ctx.Done():
			break loop
		case d, ok := <-msgs:
			if !ok {
				err = errors.New("delivery channel closed")
				break loop
			}
			attempt := attempts(d) // число неудач из заголовка x-death
			if err := process(ctx, d.Body); err != nil {
				log.Printf("fail attempt=%d: %v", attempt, err)
				_ = d.Nack(false, false) // requeue=false -> в retry.dlx
				continue
			}
			_ = d.Ack(false)
		}
	}

	// graceful shutdown: у нас prefetch=10, значит in-flight не больше 10.
	// Дождаться последней итерации цикла выше — он выйдет по ctx.Done
	// между доставками; сообщения, которые мы уже взяли, обрабатываются
	// до конца (мы не выходим из process по ctx, а завершаем начатое).
	log.Print("worker stopped cleanly")
	return err
}

Запуск из main — с signal.NotifyContext, как в Модуле 2 курса «Фундамент»:

go
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
if err := worker.Run(ctx, conn); err != nil && !errors.Is(err, context.Canceled) {
	log.Fatal(err)
}

Упражнения: посчитайте попытки честно из x-death и роутите nack в retry.1s/retry.5s/tasks.dlq в зависимости от счётчика (сейчас маршрутизацию делает топология — разберите оба варианта); добавьте параллельную обработку с сохранением порядка (курс OrderFlow). Начните с подсчёта попыток:

go
func attempts(d amqp.Delivery) int {
	if raw, ok := d.Headers["x-death"]; ok {
		if deaths, ok := raw.([]interface{}); ok && len(deaths) > 0 {
			if death, ok := deaths[0].(amqp.Table); ok {
				if cnt, ok := death["count"].(int64); ok {
					return int(cnt)
				}
			}
		}
	}
	return 0
}

Идемпотентность через Redis

go
const idemTTL = 24 * time.Hour

// Reserve помечает ключ как «в работе». false = уже обработано/обрабатывается.
func (p *Processor) Reserve(ctx context.Context, key string) (bool, error) {
	return p.redis.SetNX(ctx, "done:"+key, "in-flight", idemTTL).Result()
}

func (p *Processor) process(ctx context.Context, msgID string, body []byte) error {
	ok, err := p.Reserve(ctx, msgID)
	if err != nil {
		return err // не сумели проверить — безопаснее повторить
	}
	if !ok {
		return nil // дубль: молча подтверждаем
	}
	if err := p.handle(ctx, body); err != nil {
		// откатываем резерв, чтобы retry не «проглотил» задачу
		_ = p.redis.Del(ctx, "done:"+msgID).Err()
		return err
	}
	return nil
}

Помните окно уязвимости из Модуля 1? Если процесс упадёт после p.handle, но до подтверждения брокеру — задача придёт снова, Redis скажет «уже обработано», и работа не выполнится. Если цена ошибки высока, храните результат обработки в Redis и на дубле возвращайте его (или проверяйте факт в базе — курс Kafka покажет вариант с таблицей applied_events).

💻 Практика

1. lesson2_2: реализуйте 4 лабораторки exchange (direct/fanout/topic/headers) одним скриптом и таблицей «ключ → в какие очереди попало».
2. lesson2_3: напишите тест: продюсер без confirms + рестарт брокера (docker stop/start) — покажите потерю. Затем включите confirms и убедитесь, что публикация падает/ждёт, а не теряется молча.
3. lesson2_4: соберите цепочку отложенных задач delay.30s → tasks.main и докажите тестом, что задача вернулась через 30 ± 2 секунды.
4. lesson2_5: поднимите 3-узловой кластер, убейте rabbit1 с quorum-очередью и с classic-очередью — зафиксируйте разницу в поведении (classic умрёт вместе с узлом-мастером).
5. Промежуточный проект: «Надёжный воркер» целиком: топология + graceful shutdown + backoff + идемпотентность + тест, который роняет процесс посреди обработки и проверяет отсутствие двойного эффекта.

🧩 Quiz "Проверь себя"

✅ Итоги модуля

После прохождения ты должен уметь:

• Нарисовать AMQP-модель по памяти и объяснить, зачем разделение connection/channel/exchange/queue/binding.
• Выбрать тип exchange под задачу и написать биндинги topic-паттернов без подглядывания.
• Включить publisher confirms, объяснить разницу «Publish вернул nil» vs «получен confirm» и не потерять unroutable-сообщения (mandatory).
• Настроить prefetch по формуле и объяснить, почему prefetch=1 медленный, а безлимитный — опасный.
• Собрать DLX-цепочку с retry-таймерами и объяснить ловушку per-message TTL.
• Объяснить, что даёт quorum-очередь, сколько узлов она переживает и чем отличается от deprecated-зеркалирования.
• Написать воркер с graceful shutdown, идемпотентностью через Redis и доказать это тестом с «убийством» процесса.

Шпаргалка модуля

🔗 Куда дальше: Kafka

RabbitMQ идеально закрывает «задачи и маршрутизацию», но всё его хранение — «до ack»: историю не перечитать, догнать отставание нельзя, порядок не гарантирован. Дальше — Apache Kafka: брокер, построенный наоборот — постоянный лог с оффсетами, партициями и консьюмер-группами, где «перечитать всё с начала» — нормальная операция, а не баг. Заберём с собой привычку к confirms (она превратится в acks=all) и к идемпотентности — она понадобится ещё больше.