Паттерны очередей и интеграции

В прошлом уроке события поехали через outbox в брокер, и мы делали вид, что брокер это просто труба: положил с одного конца, достал с другого. Пора признаться, что труба устроена сложнее. Сообщения теряются, отравляют consumer-ов, устаревают, приходят не туда и не в том порядке. У этих проблем есть каталог решений, который индустрия собрала ещё в 2003 году в книге Enterprise Integration Patterns, и он до сих пор точнее описывает твою систему на RabbitMQ или Kafka, чем их собственная документация. В уроке про CQRS и шины мы выбирали между ESB, очередями и event streaming; этот урок ныряет внутрь очередей: язык каталога EIP и эксплуатационная дисциплина, ack, DLQ, TTL и сообщения-яды.

Язык EIP: каналы, сообщения, маршрутизация

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

Первая развилка каталога: сколько получателей у канала.

• Point-to-point. Одно сообщение получает ровно один потребитель, даже если читателей у очереди много. Это семантика задач: «отправь письмо» должно выполниться один раз. В RabbitMQ это обычная очередь, в Kafka партиция внутри consumer group.
• Publish-subscribe. Сообщение получает каждый подписчик. Это семантика фактов: «заказ создан» интересен и письмам, и аналитике, и антифроду, и они не должны делить одно событие. В RabbitMQ это fanout/topic exchange, раздающий копии в очереди подписчиков; в Kafka один topic, который каждая consumer group читает независимо.

Развилка вторая, из прошлого урока: что за сообщение едет. Command («сделай») с одним адресатом и ожиданием исполнения, event («случилось») с любым числом подписчиков, document (просто данные, без указания, что с ними делать). Команды просятся в point-to-point, события в pub-sub; если у тебя «событие», которое обязан услышать ровно один конкретный сервис, перечитай главу про события-команды из урока 19.

Дальше маршрутизация, три паттерна на одну идею «кому это сообщение»:

• Content-based router: выбор канала по содержимому.
• Message filter: то же, но лишнее не перенаправляется, а отбрасывается; подписчик получает только своё.
• Recipient list: отправитель сам вычисляет список адресатов и шлёт каждому копию.

И пара для работы с гранулярностью: splitter и aggregator. У aggregator-а обрати внимание на две вещи, которые всплывут на любом проектировании: ему нужно состояние (где-то копятся части, обычно по correlation id) и таймаут (что делать, если пришло 9 частей из 10: ждать вечно, отдать частичный агрегат или заявить отказ).

Последний паттерн главы решает проблему толстых сообщений: claim check, как номерок в гардеробе. Брокеры плохо переносят мегабайтные payload-ы (память, репликация, redelivery того же видео по десять раз), поэтому тяжёлое кладётся в S3, а в сообщении едет ключ.

Теперь обещанный мини-практикум. Поток афиши: «оплата пришла, чек выписан, письмо ушло, аналитика посчитана». Размечаем:

1. Вебхук оплаты падает в очередь входящих платежей: point-to-point, обработать должен один воркер.
2. Воркер публикует событие payment.completed: pub-sub, подписчики не знают друг о друге.
3. Подписчик «чеки» выписывает чек; для заказа из нескольких позиций с разным НДС внутри работает splitter (позиция на сообщение) и aggregator (собрать чек обратно).
4. Подписчик «письма» шлёт подтверждение; PDF чека едет claim check-ом: в сообщении ключ S3, не сам файл.
5. Подписчик «аналитика» фильтрует (message filter): её интересуют только оплаты дороже порога.

Пять шагов, шесть паттернов, и у каждой стрелки на схеме теперь есть имя и известные режимы отказа. В этом и польза словаря.

Гарантии доставки: ack, persistence, confirms

Спустимся на этаж ниже: что происходит с одним сообщением по дороге. Путь у него трёхзвенный, и потеряться можно на каждом звене:

1. Producer и брокер. Сообщение ушло в сокет, брокер упал до записи. Лечится publisher confirms: издатель ждёт подтверждения, а наш outbox-поллер из урока 19 не помечает событие отправленным, пока confirm не пришёл.
2. Брокер и диск. Брокер принял, но держал в памяти и перезапустился. Лечится persistent-сообщениями и durable-очередями.
3. Брокер и consumer. Самое интересное, здесь живёт ack.

Acknowledgement это контракт о моменте, когда брокер вправе забыть сообщение, и режим ack-а буквально выбирает гарантию:

• Auto-ack: сообщение считается доставленным в момент отправки consumer-у. Consumer упал посреди обработки: сообщение потеряно. Это at-most-once, годится для телеметрии и прочего, что не жалко.
• Manual ack: consumer подтверждает после успешной обработки. Упал до ack-а: брокер передоставит сообщение другому. Это at-least-once, и снова дубли: упасть после обработки, но до ack-а тоже можно. Inbox из урока 19 закрывает ровно это окно.

await channel.consume('emails', async (msg) => {
  try {
    await sendEmail(JSON.parse(msg.content.toString()));
    channel.ack(msg); // только после успеха
  } catch {
    // requeue: false, чтобы не зациклить; куда уходит сообщение,
    // решает политика DLX из следующей главы
    channel.nack(msg, false, false);
  }
});

За гарантии платим скоростью: persistent плюс confirms это запись на диск и лишний раундтрип на каждое сообщение. Поэтому решение принимается на очередь, а не на брокер: платежи едут с полными гарантиями, метрики можно и потерять.

Раз consumer теперь подтверждает сообщения, появляется вопрос: сколько неподтверждённых выдавать ему авансом? Это prefetch. Слишком большой prefetch при падении consumer-а возвращает в очередь толстую пачку, слишком маленький душит пропускную способность. А сама возможность поставить несколько одинаковых consumer-ов на одну очередь называется competing consumers и есть главный способ масштабировать обработку. Цена: порядок. Два сообщения одного заказа уйдут разным воркерам, и «оплата получена» может обработаться позже, чем «заказ отменён». Там, где порядок критичен, поток партиционируют по ключу (все события заказа 42 в одну партицию, как в Kafka) либо строят обработку так, чтобы порядок не имел значения.