День 3: Асинхронное взаимодействие и распределённые системы

🎯 Цели дня

  1. Понять гарантии доставки очередей (At-least-once) и роль ручного Ack.
  2. Освоить концепцию идемпотентности консьюмера.
  3. Разобрать паттерн Saga (хореография vs оркестрация).
  4. Поднять RabbitMQ в Docker и написать Producer/Consumer.

📖 Теоретический минимум (45 мин)

RabbitMQ: ключевые концепции

  • Durable queue — переживает перезапуск RabbitMQ.
  • Persistent delivery (DeliveryMode=2) — сообщение сохраняется на диск.
  • Manual Ack — RabbitMQ удаляет сообщение только после подтверждения консьюмером.
  • Prefetch=1 — воркер берёт не больше одного сообщения за раз (честный round-robin).

Гарантии доставки

ГарантияОписаниеКогда использовать
At-most-onceСообщение может потерятьсяЛоги, метрики
At-least-onceСообщение может быть доставлено дваждыБольшинство бизнес-задач
Exactly-onceРовно одна доставкаФинансовые транзакции (сложно!)

Важно: At-least-once — самая распространённая. Требует идемпотентности консьюмера.

Идемпотентность

Свойство операции давать одинаковый результат при повторном выполнении.

Как обеспечить:

  • Хранить event_id в БД с уникальным индексом.
  • Использовать Redis SETNX.
  • Проверять статус заказа перед обработкой.

Saga: распределённые транзакции

ACID через несколько сервисов невозможен. Saga — паттерн для обеспечения согласованности через последовательность локальных транзакций с компенсацией.

Два подхода:

  • Хореография — сервисы слушают события друг друга. Проще для 2–3 сервисов.
  • Оркестрация — центральный координатор (например, Temporal). Для сложных цепочек.

💻 Практика (2–3 часа)

Задание 1: RabbitMQ Producer/Consumer

Подними RabbitMQ через docker-compose:

yaml
version: "3.8"
services:
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest

Producer (Order Service):

go
rabbitCh.Publish("", "orders.created", false, false, amqp.Publishing{
    ContentType:  "application/json",
    DeliveryMode: amqp.Persistent, // = 2
    Body:         body,
})

Consumer (Processing Service):

go
ch.Qos(1, 0, false) // Prefetch=1
msgs, _ := ch.Consume("orders.created", "worker", false, false, false, false, nil)

for d := range msgs {
    // обработка
    d.Ack(false) // ручной ack
}

Эксперимент:

  1. Запусти Consumer, начни обработку сообщения.
  2. Оборви его (Ctrl+C) до вызова d.Ack(false).
  3. Перезапусти Consumer.
  4. Убедись, что сообщение доставлено снова.

Задание 2: Идемпотентность консьюмера

Добавь проверку перед обработкой:

go
processedEvents := make(map[string]bool)
var mu sync.RWMutex

for d := range msgs {
    mu.Lock()
    if processedEvents[d.MessageId] {
        mu.Unlock()
        d.Ack(false)
        continue
    }
    processedEvents[d.MessageId] = true
    mu.Unlock()

    // обработка
    d.Ack(false)
}

Для production: Используй БД или Redis вместо in-memory map.

🏗️ Архитектурное задание для AgentForge (1 час)

Спроектируй компенсационную логику (Saga) для сценария:

  1. AI-агент успешно создал тикет в Jira.
  2. При отправке уведомления в Telegram внешний API вернул 500.

Вопрос: Как должен выглядеть компенсирующий шаг? Ответ: Вызов API Jira для удаления созданного тикета или изменения его статуса на "Ошибка инициализации", чтобы не оставлять мусор.

Дополнительно: Опиши поток событий при хореографии:

  • order.created → Order Service
  • ticket.created → Notification Service
  • notification.failed → Order Service (компенсация)

🗣️ Топ-5 вопросов для самопроверки

  1. Как обеспечить идемпотентность консьюмера? Ответ: Хранить event_id с уникальным ограничением в БД или использовать Redis SETNX.

  2. Что будет, если консьюмер упал с паникой после бизнес-логики, но до вызова Ack? Ответ: Сообщение вернётся в очередь и будет обработано дважды. Поэтому обработка обязана быть идемпотентной.

  3. В чем разница между At-most-once, At-least-once и Exactly-once? Ответ: At-most-once — может потеряться. At-least-once — может быть дважды. Exactly-once — ровно одна доставка (сложно реализовать).

  4. Зачем нужен Prefetch=1 в RabbitMQ? Ответ: Для честного round-robin распределения. Без него быстрый воркер может забрать все сообщения, а медленный — простаивать.

  5. Чем хореография отличается от оркестрации в Saga? Ответ: Хореография — сервисы слушают события друг друга (децентрализовано). Оркестрация — центральный координатор управляет потоком (централизованно).

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

1. Как обеспечить идемпотентность консьюмера? — выберите все подходящие
2. Какая гарантия доставки самая распространённая в бизнес-задачах?
3. Чем Saga-хореография отличается от оркестрации?
4. Зачем нужен Prefetch=1?
5. Почему At-least-once требует идемпотентного консьюмера?
6. Что произойдёт, если консьюмер упал после бизнес-логики, но до d.Ack(false)?
7. Что нужно, чтобы сообщение RabbitMQ пережило перезапуск брокера? — выберите все подходящие

✅ Чек-лист дня

  • Поднял RabbitMQ в Docker
  • Написал Producer с DeliveryMode: Persistent
  • Написал Consumer с ручным Ack и Prefetch=1
  • Проверил, что сообщение не теряется при падении консьюмера
  • Реализовал идемпотентность консьюмера
  • Спроектировал Saga для AgentForge