Событийные системы: Kafka, RabbitMQ и очереди

01События, команды и брокеры

Команда просит кого-то что-то сделать: «отправь это письмо», «спиши деньги с этой карты». У неё один предполагаемый обработчик, и её можно отклонить. Событие сообщает факт, который уже произошёл: «заказ 42 оплачен». Ему всё равно, кто слушает: отреагировать могут ноль или десять потребителей. Если их путать, начинаются проблемы: «событие» с именем SendEmail привязывает producer'а к одному потребителю. Брокер сообщений — почта между ними. Producer'ы публикуют в него сообщения, он надёжно их хранит, а consumer'ы получают их в своём темпе. Если consumer лежит, сообщения ждут. Если трафик подскочил, растёт хвост очереди, а не плавится база. Цена — асинхронность: producer больше не знает, когда и выполнена ли работа вообще, поэтому нужен мониторинг хвостов и сбоев.

• Команда: повелительное наклонение, один обработчик, может провалиться («ReserveStock»).
• Событие: прошедшее время, много обработчиков, неизменяемый факт («StockReserved»).
• Брокер: надёжный буфер и маршрутизатор между producer'ами и consumer'ами.

02Kafka: topic'и, партиции, offset'ы и consumer group'ы

Kafka — распределённый лог, в который можно только дописывать. Сообщения пишутся в topic, разбитый на партиции; каждая партиция — упорядоченная последовательность, где каждое сообщение получает offset (свой порядковый номер). Сообщения с одинаковым ключом всегда попадают в одну и ту же партицию, поэтому порядок гарантирован в пределах ключа (например, ID заказа), но никогда — по всему topic'у. Consumer'ы объединяются в consumer group; Kafka назначает каждую партицию ровно одному consumer'у в группе, так что число партиций ограничивает параллелизм. Каждая группа отслеживает свои закоммиченные offset'ы, а значит, много групп (биллинг, аналитика, поиск) читают один topic независимо. Сообщения не удаляются при чтении: они хранятся определённое время или до определённого объёма либо компактируются до последнего значения по каждому ключу. Начиная с Kafka 4.0 кластеры работают на KRaft без ZooKeeper, а в Kafka 4.2 появились готовые к продакшену share group'ы для потребления в стиле очереди.

1# Create a topic: 6 partitions, 3 replicas, 7-day retention
2kafka-topics.sh --bootstrap-server localhost:9092 --create \
3  --topic orders.v1 --partitions 6 --replication-factor 3 \
4  --config retention.ms=604800000 --config min.insync.replicas=2
5 
6# Produce keyed messages (same key -> same partition -> ordered)
7kafka-console-producer.sh --bootstrap-server localhost:9092 \
8  --topic orders.v1 --property parse.key=true --property key.separator=:
9 
10# Inspect a consumer group: current offset, end offset and lag per partition
11kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
12  --describe --group billing-service

03RabbitMQ и очереди на Redis

RabbitMQ — маршрутизатор сообщений. Producer'ы публикуют в exchange, который направляет сообщения в очереди через binding'и: direct (точный routing key), topic (шаблоны вроде «order.*»), fanout (во все привязанные очереди) или headers. Consumer'ы забирают сообщения из очередей и подтверждают их; неподтверждённые сообщения доставляются повторно, а отклонённые могут уйти в dead-letter exchange. После подтверждения сообщение исчезает, так что RabbitMQ — про доставку работы, а не про хранение истории (его stream-очереди добавляют режим, похожий на лог, когда нужно повторное чтение). Он хорош для маршрутизации отдельных сообщений, приоритетов и request-reply. Очереди на Redis, такие как BullMQ (Node) или очереди Laravel с драйвером Redis, — самый лёгкий вариант: задачи с повторами, backoff, задержками, rate limit'ами, задачи по расписанию и повторяющиеся задачи — всё на том Redis, который у вас, скорее всего, уже есть. Их ограничение — надёжность и масштаб: Redis держит данные в памяти, так что настройте persistence и относитесь к очереди как к исполнителю задач, а не как к постоянному хранилищу событий.

04Гарантии доставки на практике

Любой брокер предлагает какую-то комбинацию трёх гарантий. At-most-once: consumer коммитит до обработки, так что при падении сообщение теряется, но никогда не дублируется; приемлемо для метрик, редко — для бизнес-данных. At-least-once: consumer обрабатывает, затем подтверждает; падение между этими шагами означает повторную доставку, поэтому дубликаты случаются. Это реалистичный вариант по умолчанию для большинства систем. Exactly-once: каждое сообщение влияет на результат один раз. Kafka поддерживает это внутри Kafka — с помощью идемпотентных producer'ов и транзакций, когда consumer'ы читают только закоммиченные данные, — для пайплайнов read-process-write между topic'ами Kafka. Как только побочный эффект выходит за пределы Kafka (запись в базу, письмо, списание с карты), вы снова в мире at-least-once и должны сделать consumer идемпотентным: хранить ID обработанных сообщений, использовать уникальные ограничения или upsert'ы и проектировать обработчики так, чтобы их двойной запуск ничего не менял.

• Сторона producer'а: acks=all и включённая идемпотентность (по умолчанию в современных клиентах Kafka).
• Сторона consumer'а: коммитьте offset'ы после успешной обработки, а не до.
• «Ядовитые» сообщения: повтор с backoff, затем перенос в dead-letter topic или очередь.

05Схемы, версионирование и выбор инструмента

События — это публичный контракт, который часто читают команды, с которыми вы никогда не встретитесь, поэтому относитесь к их структуре как к API. Описывайте схемы в Avro, Protobuf или JSON Schema, храните их в schema registry и применяйте правила совместимости: опциональные поля добавляйте свободно, но никогда не удаляйте и не меняйте назначение поля, на которое полагаются потребители. Для ломающих изменений публикуйте новый тип события или новую версию topic'а (orders.v2) и держите обе, пока потребители не переедут. Включайте в каждое сообщение ID события, тип, версию, таймстемп и ключ сущности. Как выбирать: Kafka — для больших потоков, долгого хранения, повторного чтения и множества consumer group'ов; RabbitMQ — для гибкой маршрутизации, распределения задач и подтверждения каждого сообщения; очереди на Redis вроде BullMQ — для фоновых задач приложения с повторами и расписанием; управляемые сервисы (SQS, SNS, Pub/Sub, managed Kafka) — когда вы не хотите сами обслуживать брокеры.

06Надёжность внутри Kafka: реплики, ISR и выбор лидера

У каждой партиции есть одна реплика-лидер, принимающая записи, и фолловеры, которые её копируют. In-sync replica set (ISR) — это лидер плюс фолловеры, отстающие не больше чем на replica.lag.time.max.ms. При acks=all продюсер получает успех, только когда запись есть у всех текущих членов ISR; min.insync.replicas задаёт, до какого размера может сжаться ISR, прежде чем записи начнут отклоняться с NotEnoughReplicas. Типичная надёжная конфигурация — replication factor 3, min.insync.replicas 2 и acks=all: один брокер может упасть без потери данных и без остановки записи. Оставьте unclean.leader.election.enable в значении по умолчанию false; если его включить, лидером может стать несинхронизированная реплика и молча потерять уже подтверждённые сообщения ради доступности. Распределяйте реплики по стойкам или зонам доступности через broker.rack, чтобы отказ одной зоны не унёс все копии.

• RF 3 + min.insync.replicas 2 + acks=all: безопасно переживает потерю одного брокера.
• Идемпотентность продюсера и acks=all — значения по умолчанию в клиенте начиная с Kafka 3.0.
• Unclean election меняет подтверждённые данные на доступность; не включайте его.

07Ребалансировка, новый протокол консьюмеров и зависания обработки

Когда консьюмеры подключаются, уходят или падают, партиции перераспределяются в ходе ребалансировки. Классический протокол мог останавливать всю группу; cooperative-sticky assignment это смягчил, а протокол нового поколения (KIP-848), общедоступный начиная с Kafka 4.0, переносит назначение на group coordinator на стороне брокера и ребалансирует инкрементально. Клиенты включают его через group.protocol=consumer; группы переводятся во время rolling upgrade. Важны два таймаута. session.timeout.ms обнаруживает мёртвого консьюмера по heartbeat. max.poll.interval.ms обнаруживает зависшего: если обработчик между вызовами poll работает дольше, консьюмер исключается, его партиции переезжают, а незакоммиченный батч обрабатывается заново в другом месте. Поэтому медленные обработчики приводят к повторной работе и штормам ребалансировок. Уменьшите max.poll.records, вынесите долгую работу в отдельную очередь или ставьте партиции на паузу на время обработки.

08Компакция, retention и replay в больших масштабах

Retention может быть по времени или размеру (политика delete) или по ключу (политика compact). Компакция хранит как минимум последнюю запись для каждого ключа и в фоне удаляет более старые, превращая топик в надёжную таблицу текущего состояния: профили клиентов, цены товаров, конфигурацию. Запись с null-значением — это tombstone: она помечает ключ на удаление и сама удаляется через delete.retention.ms, поэтому консьюмеры, отставшие сильнее, могут пропустить удаления. Сочетание обеих политик хранит последнее значение по ключу и всё равно удаляет очень старые ключи. Replay — суперсила: новая consumer group, стартующая с самого раннего offset, восстанавливает read model из истории. Планируйте это: перестроение может читать терабайты, поэтому ограничивайте скорость, выделяйте ему собственную группу и убедитесь, что retention длиннее самого плохого реалистичного времени восстановления.

1# A compacted topic holding the latest price per product
2kafka-topics.sh --bootstrap-server localhost:9092 --create \
3  --topic prices.v1 --partitions 12 --replication-factor 3 \
4  --config cleanup.policy=compact \
5  --config min.compaction.lag.ms=3600000 \
6  --config delete.retention.ms=86400000
7 
8# Rebuild a read model: new group, start from the beginning
9kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
10  --group search-indexer-v2 --topic prices.v1 \
11  --reset-offsets --to-earliest --execute

09Число партиций, перекос ключей и тюнинг пропускной способности

Партиции — единица параллелизма и порядка, поэтому выбирайте их число под параллелизм консьюмеров, который ожидаете через год-два, а не сегодня: добавление партиций позже перераспределяет ключи и ломает порядок по ключу на время изменения. Слишком много партиций тоже стоит дорого: больше открытых файлов, дольше выборы лидера и больше метаданных. Ключ с намного большим трафиком, чем у остальных, создаёт горячую партицию, которую один консьюмер вынужден обрабатывать в одиночку. Исправляйте перекос более мелким ключом (order ID вместо customer ID), если порядок важен только в пределах заказа, или «солите» известный горячий ключ на несколько подключей и сливайте результат дальше по потоку. Записи без ключа распределяются sticky partitioner, который заполняет один батч и только потом переходит к следующему. На продюсере linger.ms и batch.size меняют несколько миллисекунд latency на гораздо более высокую пропускную способность, а сжатие (lz4 или zstd) снижает нагрузку на сеть и диск.