Модуль 1. Фундамент: язык распределённых систем

Цель модуля: к концу ты должен свободно говорить на одном языке с брокерами — понимать, что такое гарантия доставки, почему «ровно один раз» — это договорённость, а не магия, и чем queue отличается от stream настолько, чтобы потом не путать семантики разных систем.

Перед началом: Go 1.22+, Docker, базовое знание context и горутин (Модули 1 и 3 курса «Go Deep Dive»). Инструменты: docker-compose, testcontainers-go, Makefile. Трудоёмкость: ~10–12 ч (теория 4 ч · практика 6 ч · квиз 1 ч).

📖 1.1. Что такое брокер и зачем он вообще

Синхронное vs асинхронное

Обычный HTTP-вызов между сервисами:

text
[Service A] --HTTP--> [Service B]
            <--200---

Проблемы:

• Coupling (связанность). A должен знать, где B, как к нему достучаться, какой у B таймаут.
• Availability (доступность). Если B упал — падает A (или вынужден сам реализовывать retry/circuit breaker).
• Backpressure (противодавление). Если B обрабатывает запрос 5 секунд, а A шлёт 1000 rps — A либо упрётся в лимиты соединений, либо B ляжет.

Брокер превращает это в:

text
[Service A] --publish--> [Broker] --deliver--> [Service B]

A больше не ждёт B. A говорит: «я сделал свою часть, вот событие». B заберёт его, когда сможет.

Когда брокер НЕ нужен

Честный инженер знает, когда он не нужен. Брокер плох, когда:

• Нужен синхронный ответ здесь и сейчас (хотя request-reply поверх брокера существует — см. урок 1.3).
• Сообщений мало и они простые — это оверкилл.
• Нужна строгая синхронная транзакционность между двумя операциями (брокер не заменит 2PC, а Saga — это уже другая история).
• Задержка критична (< 1 ms) — брокер всегда добавляет свои миллисекунды на сетевой прыжок и персистентность.

🚨 Антипаттерны на старте (запомни, будем встречать постоянно)

1. Брокер как шина всего. Один топик events для всех событий всех сервисов. Итог: никто не знает, кто что читает, эволюция схем невозможна.
2. RPC через брокер без таймаутов. Request-reply без дедлайна = зависший бизнес-процесс при падении респондера.
3. Брокер вместо БД. Хранить «текущее состояние» в очереди — гарантированная боль: очередь про доставку, не про состояние.
4. Пуш без backpressure. Продюсер шлёт столько, сколько может, консьюмер не ограничен prefetch → OOM на пике.

Go-интерфейс, который объединит весь курс

Зафиксируем абстракцию, которую будем использовать во всех модулях:

go
type Message struct {
    Key     []byte            // для партицирования или маршрутизации
    Payload []byte
    Headers map[string]string // trace-id, idempotency-key, etc.
    Topic   string
}

type Producer interface {
    Publish(ctx context.Context, msg Message) error
    Close() error
}

type Consumer interface {
    Subscribe(ctx context.Context, handler func(context.Context, Message) error) error
    Close() error
}

Этот интерфейс — наш «контракт» на весь курс. В Модуле 3 (Watermill) мы увидим, что индустрия пришла ровно к такой же абстракции.

📊 Метрики, которые нужно представлять уже сейчас

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

В Модуле 2 соберём всё в Prometheus + Grafana и привяжем алерты к SLO. Пока — просто держи эти четыре числа в голове при каждом дизайн-решении.

📖 1.2. Три гарантии доставки и почему «exactly-once» — это иллюзия

Определения

Главный секрет, который знают все сеньоры

Exactly-once между двумя независимыми системами невозможен в принципе. Это следует из задачи двух генералов и ограничений сети.

Но на практике «exactly-once» всё же делают. Как? Через идемпотентность на стороне консьюмера.

Идемпотентность — свойство операции: f(f(x)) == f(x). Применить дважды = получить тот же результат, что и при одном применении.

Пример: идемпотентный обработчик на Go

go
type IdempotentHandler struct {
    mu       sync.Mutex
    seen     map[string]struct{} // в реальности — Redis/Postgres с TTL
    delegate func(ctx context.Context, msg Message) error
}

func (h *IdempotentHandler) Handle(ctx context.Context, msg Message) error {
    id := msg.Headers["idempotency-key"]
    if id == "" {
        return errors.New("idempotency-key is required")
    }

    h.mu.Lock()
    if _, ok := h.seen[id]; ok {
        h.mu.Unlock()
        return nil // уже обработано — возвращаем успех, не падаем
    }
    h.seen[id] = struct{}{}
    h.mu.Unlock()

    return h.delegate(ctx, msg)
}

⚠️ Ловушка: race condition, о который все спотыкаются

В примере выше seen обновляется до вызова delegate. Что если delegate упал с ошибкой? Ключ уже помечен как обработанный, но по факту ничего не сделано. Сообщение ушло в никуда, retry не сработает.

Правильный паттерн (сначала работа, потом фиксация):

go
func (h *IdempotentHandler) Handle(ctx context.Context, msg Message) error {
    id := msg.Headers["idempotency-key"]

    h.mu.Lock()
    if _, ok := h.seen[id]; ok {
        h.mu.Unlock()
        return nil
    }
    h.mu.Unlock()

    // Сначала делаем бизнес-работу
    if err := h.delegate(ctx, msg); err != nil {
        return err // ключ НЕ помечен — при retry сообщение обработается снова
    }

    // Только после успеха — помечаем как обработанное
    h.mu.Lock()
    h.seen[id] = struct{}{}
    h.mu.Unlock()
    return nil
}

Но теперь другая проблема: если процесс упал между успешным delegate и записью в seen — при рестарте сообщение обработается дважды.

Решение в продакшене: писать seen в ту же транзакцию БД, что и бизнес-данные (Transactional Outbox — Модуль 2).

📖 1.3. Модели обмена: queue, pub/sub, request-reply, streaming

Четыре модели, которые покрывают 99% задач.

1. Queue (work queue)

Одно сообщение — один консьюмер. Брокер сам балансирует нагрузку между воркерами.

text
┌──> [Worker A]
[P] --> Q ─┼──> [Worker B]
           └──> [Worker C]

Когда: фоновая обработка задач, где порядок не критичен или гарантируется ключом.

2. Pub/Sub (broadcast)

Одно сообщение — все подписчики получают свою копию.

text
┌──> [Sub A]
[P] --> T ─┼──> [Sub B]
           └──> [Sub C]

Когда: уведомления, события домена, интересные многим независимым сервисам.

3. Request-Reply (RPC over MQ)

Синхронный ответ через асинхронный транспорт.

text
[A] --request--> [Broker] --request--> [B]
[A] <--reply---- [Broker] <--reply---- [B]

Реализуется через temporary reply queue (RabbitMQ) или inbox-топик (NATS). Когда: нужен ответ, но хочется сохранить decoupling на уровне инфраструктуры.

4. Streaming (event log)

Сообщения не удаляются после чтения. Каждый консьюмер читает с любого оффсета независимо.

text
[P] --> [Log: msg1, msg2, msg3, ...]
              ↑         ↑         ↑
           [C1: o=0] [C2: o=2] [C3: o=1]

Когда: event sourcing, аудит, replay событий, несколько независимых проекций.

🗺 Как модели ложатся на системы курса

Это соответствие — половина ответа на вопрос «какой брокер выбрать».

📖 1.4. Go-инструментарий курса

Чтобы каждый урок был воспроизводим, зафиксируем стек.

Docker-окружение

Весь курс живёт в docker-compose.yml. Заготовка:

yaml
services:
  rabbitmq:
    image: rabbitmq:3.13-management
    ports: ["5672:5672", "15672:15672"]
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
      interval: 5s
      retries: 10

  kafka:
    image: apache/kafka:3.7.0
    ports: ["9092:9092"]
    environment:
      KAFKA_CFG_NODE_ID: 1
      KAFKA_CFG_PROCESS_ROLES: broker,controller
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      # KRaft-конфиг без ZooKeeper — современный стандарт

  nats:
    image: nats:2.10-alpine
    ports: ["4222:4222"]
    command: ["-js"]  # JetStream сразу включён

Testcontainers — тесты без ручного запуска

Чтобы не держать брокеров поднятыми постоянно, используем testcontainers-go:

go
package broker_test

import (
    "context"
    "testing"
    "time"

    "github.com/rabbitmq/amqp091-go"
    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/wait"
)

func TestRabbitMQ_Connectivity(t *testing.T) {
    ctx := context.Background()

    req := testcontainers.ContainerRequest{
        Image:        "rabbitmq:3.13-alpine",
        ExposedPorts: []string{"5672/tcp"},
        WaitingFor:   wait.ForLog("Server startup complete"),
    }

    container, err := testcontainers.GenericContainer(ctx,
        testcontainers.GenericContainerRequest{
            ContainerRequest: req,
            Started:          true,
        })
    if err != nil {
        t.Fatal(err)
    }
    t.Cleanup(func() { _ = container.Terminate(ctx) })

    host, _ := container.Host(ctx)
    port, _ := container.MappedPort(ctx, "5672")

    conn, err := amqp091.DialTimeout(
        "amqp://guest:guest@"+host+":"+port.Port()+"/",
        5*time.Second,
    )
    if err != nil {
        t.Fatal("cannot connect:", err)
    }
    defer conn.Close()
}

Структура репозитория курса

text
course-mq/
├── module1/
│   ├── lesson1_1/      # концептуальный код
│   ├── lesson1_2/      # идемпотентный handler
│   ├── lesson1_3/      # три модели на каналах
│   └── lesson1_4/      # testcontainers setup
├── module2-rabbit/
├── module3-kafka/
├── module4-nats/
├── module5-alternatives/
├── module6-patterns/
├── module7-mvp/
├── module8-security/
├── module9-scaling/
├── docker-compose.yml
├── Makefile
└── go.work

📖 1.5. А Redis? Граница курса

В лабораторных курса Redis появится как дедуп-хранилище идемпотентных ключей (SETNX с TTL — курс RabbitMQ) — и резонно спросить: «раз Redis уже стоит, почему бы ему не быть брокером?» Короткий ответ: можно собрать очередь из Redis, но брокером он не является по назначению. Разберём границу — это любимый вопрос собеседований.

Что умеет Redis из «брокерского»

Чего не хватает честному брокеру

1. Персистентность по умолчанию. RDB теряет минуты, AOF — до секунды. Kafka/RabbitMQ флашат в рамках протокола записи; Redis — кэш с опциональной записью.
2. Нет DLQ, TTL-ретраев, dead-lettering из коробки — в RabbitMQ это DLX, здесь — паттерны поверх Streams, написанные вами.
3. Память = вся очередь. Бэклог на сутки в Kafka — дисковый вопрос, в Redis — «OOM, инцидент».
4. Backpressure слабый — prefetch как в AMQP отсутствует, лимиты считаете сами.

Когда Redis-очередь — правильный выбор

• Фоновые задачи при уже стоящем Redis — индустриальный стандарт через готовые библиотеки: Asynq (Go), Celery (Python), Sidekiq (Ruby). Это полноценные очереди задач (ретраи, backoff, scheduled, DLQ), построенные над List/Sorted Set — «надёжный воркер из курса RabbitMQ, где всё уже написано».
• Дедупликация и rate limiting (SETNX — наш случай в курсе RabbitMQ).
• Сигналинг, где потеря допустима (Pub/Sub для инвалидации кэшей).

Когда — нет

Деньги и заказы (нужен дисциплинированный ack и репликация), event sourcing и replay (нужен retention-лог), большие бэклоги (нужен disk-first storage). Это территория трёх систем этого курса.

Формулировка для собеседования: «Redis — in-memory data store, из которого очередь собирают, а не брокер по назначению. Streams дают at-least-once с группами — для фоновых задач при живом Redis это отличный выбор (Asynq). Но персистентность, DLQ, backpressure и retention там — ваша ответственность, поэтому для денег и событийных логов берут системы, для которых это контракт».

💻 Практика

1. lesson1_2: реализуйте IdempotentHandler двумя способами (фиксация до и после delegate) и напишите тест, который падает на первом и проходит на втором при ошибке delegate.
2. lesson1_3: реализуйте queue/pub-sub/streaming на каналах Go — почувствуйте семантики без брокера.
3. lesson1_4: соберите Makefile с целями up/down/test и прогоните testcontainers-тест RabbitMQ.
4. Промежуточный проект недели: «Локальный event-bus» — интерфейс Producer/Consumer поверх in-memory каналов с поддержкой idempotency-key. Он пригодится в MVP (курс OrderFlow) как fallback-драйвер.

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

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

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

• Объяснить разницу at-most/at-least/exactly-once и почему последний — всегда идемпотентность + персистентность.
• Выбрать модель обмена (queue/pub-sub/req-reply/streaming) под задачу.
• Написать идемпотентный обработчик на Go и показать его «окно уязвимости».
• Назвать четыре золотые метрики брокер-взаимодействия и что они означают.
• Поднять брокера в интеграционном тесте через testcontainers-go.
• Распознать четыре стартовых антипаттерна в чужом дизайне.
• Объяснить, когда Redis-очередь (List/Streams/Asynq) легитимна, а когда нужен настоящий брокер.

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

🔗 Мостик к Модулю 2

Язык освоен: гарантии, идемпотентность, четыре золотых метрики. Дальше — Модуль 2: паттерны и боевые шрамы — цена гарантий порядка, backpressure, poison pill и retry-стратегии, Outbox, Saga, мониторинг и graceful shutdown. Это слой, где абстракции Модуля 1 превращаются в проектные решения, общие для любого брокера. А конкретика по инструментам — в отдельных курсах трека: Kafka, RabbitMQ и Redis.