Брокеры сообщений и асинхронность

8.1Общая теория

Базис, на котором стоят и Kafka, и RabbitMQ, и любой самописный «воркер поверх таблицы». Пока не видно, где именно рвётся, разговор про acks и prefetch сводится к перечню настроек без смысла.

Сначала — словарь: кто здесь кто

Дальше на каждом шагу будут «продюсер», «консьюмер», «ack» и «гарантия доставки». Все четыре понятия простые, но договориться о них надо до разговора про надёжность, иначе остальной текст рассыпается на непонятные слова.

1. Сообщение — это данные плюс конверт

Сообщением называют порцию байтов, которую одна программа передаёт другой не напрямую, а через посредника. У сообщения есть тело (payload: сами данные, обычно JSON или бинарный формат) и заголовки, служебные поля рядом с телом: идентификатор, тип события, время, данные для трассировки. По заголовкам получатель решает, что делать с сообщением, не разбирая тело.

2. Продюсер пишет, консьюмер читает

Продюсер (producer, издатель, отправитель) отправляет сообщение. Консьюмер (consumer, потребитель, подписчик) его получает и обрабатывает. Один и тот же сервис обычно и то и другое сразу: читает из одной очереди, пишет в другую. Русские кальки «производитель / потребитель» звучат криво, поэтому и в разговоре, и на собесе почти всегда говорят по-английски.

3. Брокер — это посредник с диском

Брокер сообщений живёт на отдельном сервере (обычно на кластере серверов): принимает сообщения от продюсеров, кладёт их на диск и отдаёт консьюмерам. Диск тут главное: из-за него продюсер и консьюмер не обязаны работать одновременно. Продюсер отдал сообщение и ушёл, консьюмер придёт через час и заберёт. Kafka, RabbitMQ, NATS, SQS — всё это брокеры; устроены они по-разному, но роль одна.

4. Подтверждение (ack) — момент, когда сообщение считается обработанным

Консьюмер шлёт брокеру ack (от acknowledgement, «подтверждение»), и смысл у него такой: «это сообщение можно мне больше не показывать». Пока ack не пришёл, брокер считает сообщение невыполненной работой и рано или поздно отдаст его снова — этому же консьюмеру или другому. Обратный сигнал, nack (negative ack, «не смог»), говорит «верни в очередь» или «убери в свалку битых сообщений».

Дальше всё упирается в одно: подтверждение и обработка происходят в разные моменты, и между ними всегда есть щель, в которую можно упасть. Если подтвердить раньше, чем сделать, падение потеряет работу. Если позже — после падения работу сделают дважды. Третьего варианта нет, и ниже станет ясно, почему его не бывает в принципе.

Зачем вообще асинхронность

При синхронном вызове мало дождаться ответа: вызываемый сервис обязан быть жив прямо сейчас. Такая связанность называется temporal coupling (связанность во времени), и именно она превращает набор микросервисов в распределённый монолит: падает почта — не создаётся заказ.

Брокер даёт три вещи, и стоит проговаривать их именно так:

• Развязка (decoupling). Продюсер знает только имя топика или exchange. Он не знает, сколько у события подписчиков, живы ли они и что они с ним сделают. Чтобы добавить четвёртого потребителя, код меняют только у него.
• Сглаживание пиков (load levelling). Брокер работает буфером. Продюсер может выдавать 20 000 сообщений в секунду в пиковую минуту, а консьюмер спокойно переваривать 3 000/с следующие полчаса. Без буфера пришлось бы держать железо под пик.
• Надёжность. Сообщение лежит на диске и переживает падение потребителя. Синхронный HTTP-запрос при 500-й ошибке теряется целиком — если только вызывающий сам не построил себе очередь ретраев, то есть тот же брокер, только плохой.

Push vs pull

Вопрос в том, кто начинает передачу: брокер или потребитель. В модели push («толкать») брокер сам шлёт сообщение, как только оно появилось. В pull («тянуть») консьюмер сам приходит и спрашивает: «есть что-нибудь?». Это не деталь реализации: от неё зависит, где именно копится работа, когда потребитель не успевает.

Через backpressure (обратное давление) медленный получатель заставляет быстрого отправителя сбавить темп. Бытовой пример: раздача на кухне ресторана. Если повар ставит тарелки на стойку быстрее, чем официанты их разносят, стойка переполнится, и дальше возможны только два исхода: либо повар видит, что места нет, и притормаживает (обратное давление сработало), либо тарелки летят на пол (потери). В push тормозить отправителя приходится специально, в pull это выходит само собой: не пришёл за сообщением — оно просто осталось лежать у брокера.

Чистого push и чистого pull почти не бывает. RabbitMQ толкает, но с кредитным окном (basic.qos(prefetch)): брокер не пошлёт больше N неподтверждённых сообщений. Kafka тянет, но с long polling: запрос висит на брокере до fetch.max.wait.ms или пока не накопится fetch.min.bytes, поэтому пустых опросов почти нет и задержка остаётся единицами миллисекунд.

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

Три уровня, и все три называют по-английски, потому что русские варианты не прижились. Читаются они буквально: at-most-once значит «не более одного раза», at-least-once — «не менее одного раза», exactly-once — «ровно один раз». Формально речь не о «доставке», а о том, сколько раз может проявиться эффект обработки: не сколько раз сообщение прилетело по сети, а сколько раз списались деньги или ушло письмо.

Что физически значит «ровно один раз», видно на бытовом примере. Курьер привёз посылку и должен отметить в системе, что доставил. Он позвонил в дверь, отдал коробку — и в этот момент у него сел телефон, отметка не ушла. Диспетчер видит: заявка не закрыта. Что произошло, он не знает: курьер вообще не доехал или доехал, отдал и не отчитался. Вариантов у диспетчера ровно два: послать вторую посылку (получатель получит две) или не посылать (может остаться ни с чем). Способа узнать правду у него нет, и дело не в плохой системе: оборвался единственный канал связи. Между брокером и консьюмером такое случается десятки раз в сутки.

Почему exactly-once невозможен: проблема двух генералов

Классическая формулировка: две армии на холмах по обе стороны от долины, в долине противник. Атака удастся только при одновременном ударе. Гонцы ходят через долину, и их могут перехватить. Доказано, что никакого протокола с конечным числом сообщений, гарантирующего согласованное решение, не существует. Доказательство от противного: возьмём кратчайший корректный протокол; его последнее сообщение может потеряться, но раз протокол корректен и без него, это сообщение лишнее, и мы получили протокол короче. Повторяем, пока сообщений не останется вовсе, и приходим к противоречию.

Прикладной перевод: отправитель никогда не может достоверно узнать, был ли получен и обработан его последний запрос. Не получив ответа, он вынужден выбирать между «повторить» (риск дубликата) и «не повторять» (риск потери). Брокер этот выбор не отменяет, а лишь берёт на себя. Сюда же относится FLP-теорема (Fischer, Lynch, Paterson, 1985): в асинхронной системе, где может отказать даже один узел, детерминированный консенсус за конечное время невозможен.

Идемпотентность консьюмера

Операция идемпотентна, если даёт один и тот же результат, сколько бы раз её ни повторили: один раз, три раза, сто. Так устроена кнопка вызова лифта. Нажал её один раз или продолбил пятнадцать — лифт приедет один раз, состояние «лифт вызван» от повторов не меняется. А вот кнопка «взять талончик» в очереди неидемпотентна: каждое нажатие выдаёт новый номер. Слово латинское: idem — «тот же», potens — «имеющий силу»; формально f(f(x)) = f(x).

Раз дубликаты неизбежны, обработчик устраивают так, чтобы повторная обработка того же сообщения не меняла результат. Путей два, и они принципиально разные.

Путь первый: сделать саму операцию идемпотентной по природе. Он лучше, потому что ничего не надо хранить. UPDATE users SET status='active' WHERE id=$1 идемпотентен, а UPDATE accounts SET balance = balance - 100 — нет. Приёмы: UPSERT по естественному ключу, установка абсолютного значения вместо инкремента, оптимистическая блокировка по версии (WHERE version < $2), условная вставка ON CONFLICT DO NOTHING.

Путь второй: явная дедупликация по ключу. Дедупликация означает «вести список уже сделанного и сверяться с ним»: у каждого сообщения есть ключ, обработчик перед работой смотрит, нет ли этого ключа в списке, и если есть, молча пропускает сообщение. Как журнал на вахте: фамилия уже записана — второй раз не пускаем. Этот путь нужен, когда операция принципиально неидемпотентна (списание денег, отправка письма, вызов внешнего API). Ключ дедупликации выбирают в таком порядке:

• Бизнес-ключ: payment_id, order_id + event_type, aggregate_id + version. Самый надёжный вариант: такой ключ переживает и переотправку продюсером, и повторную генерацию события.
• Message ID от продюсера: UUID, который кладут в заголовок при первой отправке. Генерируют его один раз, а не на каждом ретрае.
• Координаты в брокере (topic:partition:offset) работают только внутри одной партиции и ломаются при перекладывании через retry-топик. Это крайний вариант.

-- Таблица дедупликации. Хватает одного PRIMARY KEY.
CREATE TABLE processed_messages (
    consumer      text        NOT NULL,   -- имя группы: один и тот же msg могут есть разные консьюмеры
    message_key   text        NOT NULL,   -- бизнес-ключ или message_id
    processed_at  timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (consumer, message_key)
);

-- Чистка: без неё таблица растёт вечно. Старое удаляют пачками по этому индексу:
-- секционировать по processed_at нельзя, ключ таблицы обязан её включать.
CREATE INDEX ON processed_messages (processed_at);

// Дедупликация и бизнес-логика обязаны быть в одной транзакции.
// Иначе получаем ту же проблему двойной записи, только в миниатюре.
func (h *Handler) Handle(ctx context.Context, m Message) error {
    tx, err := h.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback() // no-op после успешного Commit

    // ON CONFLICT DO NOTHING: если строка уже есть, RowsAffected == 0
    res, err := tx.ExecContext(ctx,
        `INSERT INTO processed_messages (consumer, message_key)
         VALUES ($1, $2) ON CONFLICT DO NOTHING`,
        h.group, m.Key)
    if err != nil {
        return err
    }
    if n, _ := res.RowsAffected(); n == 0 {
        // Уже обрабатывали. Коммитить нечего, просто подтверждаем брокеру.
        return nil
    }

    if err := h.apply(ctx, tx, m); err != nil { // вся бизнес-логика через tx
        return err
    }
    return tx.Commit()
}

Персистентность

Персистентность (от persistent, «стойкий, сохраняющийся») гарантирует, что сообщение переживёт перезапуск брокера. Для этого сообщение мало принять: его записывают туда, откуда оно достанется после включения. Обратный режим называется transient: сообщение живёт только в оперативной памяти брокера и исчезает вместе с процессом.

Тонкость, на которой ловят: «записано на диск» не равно «не потеряется». Между write() и физическим носителем стоит page cache, область оперативной памяти, где ядро ОС держит данные, ещё не сброшенные на диск. Вызов write() возвращает успех сразу, как только байты легли в этот кэш, а на пластину или флеш они попадут когда-нибудь потом. Заставить сбросить их прямо сейчас умеет только fsync; без него отключение питания стирает всё, что ещё лежало в кэше.

• Kafka сознательно не делает fsync на каждую запись: пишет в page cache, а сбрасывает по log.flush.interval.messages / .ms (по умолчанию практически «никогда, отдано ОС»). Надёжность держится не на диске одного узла, а на репликации: acks=all + min.insync.replicas=2 при replication.factor=3. Данные потеряются, только если одновременно упадут несколько брокеров.
• RabbitMQ идёт от сообщения: очередь должна быть durable, сообщение помечено delivery_mode=2 (persistent), и обязательно нужны publisher confirms, иначе basic.publish работает как fire-and-forget, и брокер вправе молча потерять сообщение. Quorum queues добавляют репликацию через Raft.

За персистентность платят задержкой и пропускной способностью. Поэтому в проде встречается смешанный режим: критичные события (платежи) идут persistent с подтверждениями, а метрики и телеметрия — transient, потому что терять их не страшно, а платить за них полной надёжностью дорого.

Ordering: где порядок есть, а где его нет

Под ordering понимают порядок сообщений: гарантию, что консьюмер увидит события в той же последовательности, в какой их отправлял продюсер. Звучит как мелочь, но без неё «заказ отменён» иногда приходит раньше «заказ создан», и обработчик не понимает, что ему отменять.

Общее правило: порядок гарантируется только внутри одной единицы упорядочивания. Так называют минимальный кусок системы, где сообщения гарантированно выстроены в одну линию. Таких единиц в системе много, и между собой они никак не синхронизированы. В Kafka эту роль играет партиция (один физический файл-лог внутри топика, подробно в 8.2), в RabbitMQ — очередь при единственном консьюмере и без requeue, в JetStream — стрим (или subject-фильтр внутри него). Глобального порядка «по топику» нет ни у кого, иначе не было бы горизонтального масштабирования.

Split brain у кластеров брокеров

Сеть делится на две части (network partition). Обе половины видят, что «остальные умерли», обе выбирают себе лидера и обе начинают принимать записи. Когда сеть починят, останутся две расходящиеся истории, и корректно слить их нельзя: сообщения либо теряются, либо дублируются, либо оказываются в разном порядке у разных потребителей.

Защищаются все одинаково: через кворум большинства. Кворумом называют минимальное число узлов, которые должны согласиться, чтобы решение считалось принятым; «большинство» значит N/2+1. Смысл в том, что два любых большинства из одного набора узлов обязательно пересекаются хотя бы в одном узле, а он не проголосует за два разных решения сразу. Поэтому две половины разорванной сети не могут обе набрать кворум, и второй лидер не появляется. Реализовано это у всех по-разному:

• Kafka. Раньше контроллер и метаданные жили в ZooKeeper (кворум ZAB), с KIP-500 переехали в саму Kafka, в KRaft (Raft-кворум контроллеров). Production-ready с 3.3, миграция с ZooKeeper готова для продакшена с 3.6, а в Kafka 4.0 (2025) ZooKeeper удалён совсем. Данные страхует unclean.leader.election.enable=false (так по умолчанию): если ни одна реплика из ISR (in-sync replicas, копии партиции, которые не отстали от лидера; подробно в 8.2) не доступна, партиция становится недоступной, но не расходящейся. Это осознанный выбор CP в терминах CAP.
• RabbitMQ. Классические зеркалируемые очереди (mirrored) split brain переживали плохо: при автоматическом слиянии одна половина просто откатывалась, теряя сообщения. Поэтому они объявлены устаревшими в 3.9 и удалены в 4.0, а штатным вариантом стали quorum queues на Raft: пишет только большинство, меньшинство отказывает в записи. Для кластера настраивают cluster_partition_handling = pause_minority или autoheal.
• NATS JetStream ведёт отдельную Raft-группу на каждый стрим, R=3 или R=5.

Очередь vs топик

Базовых моделей обмена две, и различаются они тем, сколько потребителей увидит одно сообщение.

CREATE TABLE processed_messages (
    consumer      text        NOT NULL,   -- имя группы: одно сообщение едят несколько консьюмеров
    message_key   text        NOT NULL,
    processed_at  timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (consumer, message_key)
);
-- PARTITION BY RANGE (processed_at) здесь не выйдет: первичный ключ секционированной
-- таблицы обязан включать processed_at. Старое чистят DELETE пачками по индексу.
CREATE INDEX ON processed_messages (processed_at);

8.2Kafka

Kafka устроена не как очередь, а как распределённый упорядоченный лог с указателями чтения. Из этой фразы выводятся почти все её особенности (и почти все грабли): сообщение не удаляется после прочтения, порядок живёт внутри партиции, а «кто что прочитал» Kafka записывает отдельно, в служебный топик.

Сначала — что такое лог и чем он не очередь

Большинство недоразумений вокруг Kafka растёт из одного места: человек читает «Kafka — брокер сообщений» и достраивает привычную картинку «положил, забрал, исчезло». В Kafka сообщение никуда не исчезает. Разберём это на трёх шагах, без них дальше будет каша.

1. Лог — это файл, в который только дописывают

Слово лог здесь означает не «логи приложения», а структуру данных: последовательность записей, к которой можно только дописать в конец. Изменить или удалить запись в середине нельзя — такой операции просто нет. По-английски это называется append-only log, «журнал только на дозапись».

Так устроен бумажный кассовый журнал. Каждая операция пишется новой строкой, строки нумеруются подряд, вымарывать написанное запрещено. Ошибся — пишешь новую строку «отмена операции №17». Такой журнал можно перечитать с начала и восстановить всю историю: из него ничего не пропало.

2. Читатель — это закладка, а не потребитель

В очереди чтение сообщение расходует: забрал — его больше нет. В логе чтение не меняет ничего. Каждый читатель держит собственный номер строки, на которой он остановился; этот номер и называется offset («смещение»). Прочитать запись означает взять строку под своим номером и сдвинуть номер на единицу вперёд.

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

3. Чем это отличается от очереди — практически

В эксплуатации это значит, что Kafka не знает, «сколько сообщений в очереди». Она знает, где конец лога и где чья закладка. Разница между ними называется lag, и это единственная метрика, которая тут вообще имеет смысл.

Архитектура: из чего это собрано

Кластер состоит из нескольких брокеров (процессов JVM), у каждого свой набор дисков. Топик остаётся логическим именем, физического воплощения у него нет. Хранятся данные в партициях: партиция занимает каталог на диске конкретного брокера, внутри лежит append-only лог. Топик с 12 партициями даёт 12 независимых логов, размазанных по кластеру.

Партиция нарезана на сегменты, каждый из трёх файлов:

• 00000000000000012345.log хранит сами записи, батчами, ровно в том виде, в каком их прислал продюсер;
• .index — разреженный индекс «относительный offset → байтовое смещение в .log» (запись примерно раз в index.interval.bytes = 4 КБ);
• .timeindex переводит «timestamp → offset» и нужен для retention.ms и для offsetsForTimes() при перемотке по времени.

Число в имени файла, base offset, совпадает с offset первой записи сегмента. Пишут всегда только в активный сегмент; он закрывается и создаётся новый по segment.bytes (1 ГБ по умолчанию) или segment.ms (7 суток). Удаление и компакция работают сегментами, а не записями, и активный сегмент не трогают никогда. При низком трафике «протухшее» сообщение может пролежать сильно дольше retention.ms.

Offset служит номером записи внутри партиции: номера монотонно растут, назначает их лидер. Он не глобальный: offset 100 в P0 и offset 100 в P1 не связаны ничем. Запись по offset брокер ищет так: бинарным поиском по именам сегментов, затем по .index, затем линейным досканированием внутри 4 КБ. Выходит, что чтение с произвольного offset стоит примерно столько же, сколько чтение с конца.

У каждой партиции есть лидер и фолловеры (replication.factor). Весь трафик записи и (по умолчанию) чтения идёт через лидера; фолловеры непрерывно тянут данные тем же fetch-протоколом, что и обычные консьюмеры. В ISR (in-sync replicas) входят реплики, которые не отстали больше чем на replica.lag.time.max.ms (30 с). High watermark отмечает максимальный offset, реплицированный на все реплики из ISR; консьюмеры видят записи только до HW, иначе после смены лидера они могли бы прочитать то, чего в новом лидере нет.

Координацию (кто лидер, где партиции, кто в ISR) раньше вёл ZooKeeper. После KIP-500 метаданные живут в самой Kafka, в режиме KRaft: выделенные controller-узлы держат Raft-кворум и ведут собственный лог метаданных. Production-ready с 3.3, миграция с 3.6, в Kafka 4.0 (2025) ZooKeeper удалён полностью. На практике failover контроллера стал на порядок быстрее, а кластер держит миллионы партиций вместо пары сотен тысяч.

Почему Kafka быстрая

Вопрос любят задавать, потому что ответ проверяет понимание работы с диском и сетью, а не знание Kafka. Четыре причины, по убыванию значимости.

1. Последовательная запись append-only. Нет обновлений на месте, нет B-дерева, нет случайных seek. Последовательная запись на обычный HDD даёт сотни МБ/с, что сопоставимо со случайным доступом к памяти. Kafka нарочно спроектирована так, чтобы обращаться к диску только последовательно.
2. Page cache вместо своего кэша. Kafka не держит данные в JVM heap, а пишет и читает через страничный кэш ядра. GC от этого не страдает, кэш переживает перезапуск брокера, а «горячие» читатели (те, кто у хвоста лога) вообще не касаются диска. Отсюда практическое правило: брокеру дают маленький heap (6–8 ГБ), а всю остальную память отдают ОС.
3. Zero-copy. Консьюмеру данные отдаются через sendfile(2): page cache → сокет, без копирования в user space и обратно. Это экономит два копирования и два переключения контекста. Оговорка: zero-copy отваливается, если включён TLS (данные надо шифровать в user space) или если брокеру приходится конвертировать формат записи под старого клиента. Отсюда и ответ на вопрос «почему включение TLS уронило пропускную способность вдвое».
4. Батчинг и единый формат. Продюсер собирает record batch, сжимает его целиком (а не каждое сообщение), брокер кладёт батч на диск как есть и как есть отдаёт консьюмеру. Распаковки и перепаковки на брокере нет, и стоимость сетевого и дискового вызова размазывается на тысячи сообщений.

Producer: путь одной записи

Send() в Kafka-клиенте не отправляет ничего. Он сериализует ключ и значение, спрашивает у partitioner номер партиции, кладёт запись в аккумулятор (буфер батчей, по батчу на партицию) и возвращает управление. Отдельный поток-отправитель забирает готовые батчи, группирует их по брокерам-лидерам и шлёт один запрос на брокер.

acks и min.insync.replicas

Идемпотентный продюсер

Продюсер отправил батч, брокер записал, ответ потерялся в сети, продюсер повторил, и в логе две одинаковые записи. Идемпотентность (enable.idempotence=true, включена по умолчанию с Kafka 3.0) это лечит: продюсер при старте получает PID (producer id) и ведёт sequence number отдельно на каждую партицию. Брокер помнит последний принятый номер по паре (PID, партиция): пришёл тот же номер — DuplicateSequenceNumber, батч молча отбрасывается с успешным ответом; пришёл номер с разрывом — OutOfOrderSequenceException.

Побочный эффект, за который её и любят: идемпотентность чинит порядок при ретраях. Без неё при max.in.flight.requests.per.connection > 1 повторно отправленный первый батч приедет после второго. С ней брокер по sequence number отклоняет батч с разрывом в номерах, продюсер отправляет заново по порядку, и до 5 in-flight безопасно. Границы гарантии: одна сессия продюсера, одна партиция. Перезапустили процесс — новый PID, дедупликация не работает. На другую партицию она тоже не распространяется.

Ключ, партиция и порядок

// Ключ есть   -> partition = murmur2(key) % numPartitions  (детерминированно)
// Ключа нет   -> Java-клиент: sticky partitioner, льёт в одну партицию, пока не закроется
//                батч, и переключается на следующую. kafka-go Murmur2Balancer берёт
//                для записи без ключа случайную партицию.
w := &kafka.Writer{
    Addr:     kafka.TCP("kafka-1:9092", "kafka-2:9092"),
    Topic:    "orders",
    Balancer: &kafka.Murmur2Balancer{}, // как в Java-клиенте. kafka.Hash{} считает FNV-1a
                                      // и совпадает с Sarama, но не с Java
    RequiredAcks: kafka.RequireAll,   // acks=all
    Async:    false,                  // Async=true = потеря при падении процесса
    BatchTimeout: 10 * time.Millisecond, // аналог linger.ms
    BatchSize:    100,
    Compression:  kafka.Lz4,
}
err := w.WriteMessages(ctx, kafka.Message{
    Key:   []byte(orderID),           // всё про один заказ -> одна партиция -> порядок
    Value: payload,
    Headers: []kafka.Header{
        {Key: "event_id",    Value: []byte(eventID)},   // для дедупликации у потребителя
        {Key: "traceparent", Value: []byte(traceID)},   // трейсинг сквозь брокер
    },
})

Батчинг, linger.ms и компрессия

batch.size (16 КБ) ограничивает батч на партицию по байтам. linger.ms (с Kafka 4.0 по умолчанию 5 мс, раньше 0) говорит, сколько ждать доукомплектования батча перед отправкой. Даже ноль не означает «без батчинга»: батч всё равно собирается за время, пока предыдущий запрос в полёте. Дешевле всего поднять пропускную способность в разы, увеличив linger.ms до 5–20 мс: батчи крупнее, компрессия эффективнее, запросов меньше. Платим фиксированной добавкой к задержке.

Сжимается батч целиком, и чем он крупнее, тем выгоднее компрессия. На практике lz4 почти всегда правильный дефолт (быстрый, приличная степень); zstd (с Kafka 2.1) жмёт заметно лучше при чуть большем CPU и хорош для JSON и дорогого трафика между зонами; gzip самый медленный, брать его не стоит; snappy остался легаси-компромиссом. Сжатый батч так и лежит на диске и в том же виде уходит консьюмеру: брокер его не распаковывает (кроме случаев, когда включена валидация или конвертация формата).

Consumer groups и ребалансировка

Consumer group (группа потребителей) объединяет несколько процессов, которые вместе читают один топик и делят его партиции: каждая партиция достаётся ровно одному участнику группы. Группа распараллеливает обработку, не ломая порядок внутри партиции. Ребалансировкой (rebalance) называют перераздачу партиций между участниками: кто-то пришёл, кто-то ушёл или перестал отвечать, и надо заново решить, кто какие партиции читает. Похоже на официантов, которые делят столы в зале: вышел один покурить, его столы срочно раздают остальным, и на время передела обслуживание встаёт.

Консьюмеры объединяются в группу по group.id. Один брокер становится для этой группы group coordinator (выбирается по хешу group.id): ведёт список членов, раздаёт партиции и хранит offset. А вот считает раздачу лидер группы, один из консьюмеров: координатор присылает ему список всех членов и подписок, тот раскладывает партиции по выбранной стратегии и возвращает план. Логика назначения остаётся в клиенте, и её можно менять без обновления кластера.

Ребалансировка запускается, когда меняется состав группы или набор партиций:

• новый консьюмер вызвал Subscribe и прислал JoinGroup;
• консьюмер вышел штатно (Close шлёт LeaveGroup);
• консьюмер перестал слать heartbeat дольше session.timeout.ms (45 с в новых версиях, heartbeat идёт из отдельного потока каждые heartbeat.interval.ms = 3 с);
• консьюмер не вызвал poll() дольше max.poll.interval.ms (5 минут) — такой считается зависшим, даже если heartbeat идёт;
• в топик добавили партиции или изменилась подписка по регулярке.

Стратегии назначения

Коммит offset

Committed offset равен offset следующей записи, которую надо прочитать, то есть «последняя обработанная + 1». Хранится он в служебном топике __consumer_offsets (50 партиций, cleanup.policy=compact) под ключом (group.id, topic, partition). Компакция нужна, чтобы топик не рос бесконечно: по каждому ключу важно только последнее значение.

Авто-коммит (enable.auto.commit=true, интервал 5 с) коммитит не по таймеру в фоне, а внутри вызова poll(): если с прошлого коммита прошло больше интервала, коммитится всё, что было выдано предыдущим poll. Пока цикл синхронный, эти записи к тому моменту уже обработаны, и выходит тот же at-least-once. Опасно, когда обработку отдают в другие горутины: тогда коммит подтверждает записи, обработку которых ты ещё не завершил, и падение в этом окне теряет сообщение навсегда. В librdkafka и в kafka-go с ReadMessage offset запоминается уже при выдаче, так что коммит может обогнать обработку и без всяких горутин.

// segmentio/kafka-go: ручной коммит после успешной обработки.
r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:  []string{"kafka-1:9092"},
    GroupID:  "billing",              // GroupID включает управление группой
    Topic:    "orders",
    MinBytes: 10e3,                   // fetch.min.bytes
    MaxBytes: 10e6,
    MaxWait:  500 * time.Millisecond, // fetch.max.wait.ms
    // CommitInterval = 0: CommitMessages коммитит синхронно
})
defer r.Close()

for {
    // FetchMessage, а не ReadMessage: ReadMessage коммитит сам
    m, err := r.FetchMessage(ctx)
    if err != nil {
        return err                    // ctx отменён или reader закрыт
    }
    if err := handle(ctx, m); err != nil {
        // continue здесь потерял бы сообщение: коммит следующего сдвинет
        // offset и за него. Откладываем в retry-топик, потом коммитим.
        if err := toRetry(ctx, m); err != nil {
            return err                // не отложили — выходим без коммита
        }
    }
    if err := r.CommitMessages(ctx, m); err != nil {
        slog.Error("commit", "err", err)
    }
}

Retention и log compaction

Retention («удержание», срок хранения) определяет, сколько времени или байт Kafka держит записи, прежде чем выбросить. Log compaction («уплотнение лога») убирает иначе: вместо «выбросить старое по времени» брокер выбрасывает устаревшие версии, оставляя по каждому ключу только последнюю запись. Аналогия из магазина: retention говорит «храним чеки за три месяца, дальше в мусор», а compaction — «по каждому товару держим только актуальный ценник, старые выбрасываем, сколько бы им ни было лет».

Общее для обеих политик: Kafka удаляет данные по cleanup.policy, а не по факту прочтения. Никто не прочитал — всё равно удалит по сроку; все прочитали — всё равно будет хранить до срока.

• delete (дефолт): целые сегменты удаляются, когда все их записи старше retention.ms (7 суток) либо когда суммарный размер партиции превысил retention.bytes. Обе границы работают одновременно: что наступит раньше. Считать надо на партицию, а не на топик: retention.bytes=1GB при 24 партициях дают 24 ГБ на топик, и ещё умножить на replication factor.
• compact — размер ограничен не временем, а числом уникальных ключей: для каждого ключа хранится последнее значение. Топик превращается из журнала изменений в материализованный снимок состояния, который можно перечитать с нуля и восстановить всю таблицу.
• compact,delete включает обе: последнее значение по ключу, но не старше retention. Так делают, например, «состояние за последние 30 дней».

Consumer lag

Lag («отставание») показывает, сколько записей в партиции уже написано, но этой группой ещё не обработано. Формально lag = log end offset − committed offset для каждой пары (группа, партиция): конец лога минус закладка. Это как стопка непрочитанных писем в ящике: важно не то, что там что-то лежит, а растёт она или тает.

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

Чем мерить

• kafka-consumer-groups.sh --describe --group billing — быстро и руками, показывает лаг по каждой партиции.
• kafka-exporter отдаёт в Prometheus метрику kafka_consumergroup_lag с разбивкой по партициям (у kafka-lag-exporter она называется kafka_consumergroup_group_lag). Алертить надо на производную (лаг растёт N минут подряд), а не на абсолютное значение: 50 000 при потоке 100k/с дают всего полсекунды отставания, и это нормально.
• Lag в единицах времени лучше всего подходит для SLO: «отстаём на 8 секунд» понятно бизнесу, а «отстаём на 400 000 сообщений» нет. Считается по .timeindex или разностью timestamp последней обработанной записи и текущего времени.
• Burrow (LinkedIn) смотрит не на сам лаг, а на его тренд, и выдаёт статус OK/WARN/ERR без ручных порогов.

Лаг растёт: что смотреть по порядку

Ретраи и DLQ

В Kafka нет ни встроенного requeue, ни встроенного DLQ. Нельзя «вернуть сообщение в очередь»: указатель всего один, offset, и он либо двигается, либо нет. Остаются два примитивных подхода, и оба плохи:

• Не коммитить и читать заново. Партиция встаёт колом на одном ядовитом сообщении (head-of-line blocking): за ним копятся тысячи корректных, а мы в бесконечном цикле падаем на одном. Вдобавок так быстро превышается max.poll.interval.ms, и начинается ребалансировка.
• Ретраить в цикле внутри обработчика. То же самое: пока спим между попытками, не вызываем poll(). Работает только при коротком общем бюджете (несколько секунд) и обязательно с pause() партиции и дальнейшими вызовами poll().

В проде делают цепочку retry-топиков с возрастающей задержкой и финальный DLQ.

# Топология
orders                 -> основной консьюмер
orders.retry.5s        -> консьюмер, который перед обработкой спит до (produced_at + 5s)
orders.retry.1m
orders.retry.10m
orders.DLQ             -> никто не читает автоматически; читают глазами и руками

# Заголовки, которые надо переносить вместе с сообщением:
x-original-topic       # откуда пришло изначально
x-retry-count          # сколько попыток уже было
x-first-failed-at      # когда сломалось впервые
x-last-error           # тип и текст последней ошибки
x-original-offset      # координаты оригинала, чтобы найти его в логе

// Консьюмер retry-топика спит до времени готовности сообщения.
// У всех сообщений в orders.retry.1m одинаковая задержка,
// поэтому топик упорядочен по времени готовности и достаточно
// подождать голову: приоритетный разбор не нужен.
func (c *RetryConsumer) run(ctx context.Context) error {
    for {
        m, err := c.reader.FetchMessage(ctx)
        if err != nil {
            return err
        }
        ready := m.Time.Add(c.delay)
        if d := time.Until(ready); d > 0 {
            // kafka-go держит членство в группе фоновым heartbeat, спать можно.
            // Java-клиенту на это время нужны pause() и дальнейшие вызовы poll().
            select {
            case <-time.After(d):
            case <-ctx.Done():
                return ctx.Err()
            }
        }
        var perr error
        switch err := c.handle(ctx, m); {
        case err == nil:
            // успех
        case errors.Is(err, ErrPermanent) || retryCount(m) >= c.maxRetries:
            perr = c.produce(ctx, c.dlqTopic, m) // в DLQ без дальнейших попыток
        default:
            perr = c.produce(ctx, c.nextTopic, m) // на следующую ступень задержки
        }
        if perr != nil {
            return perr // не переложили — не коммитим, прочитаем снова
        }
        if err := c.reader.CommitMessages(ctx, m); err != nil {
            return err
        }
    }
}

Транзакции и exactly-once

Транзакции Kafka закрывают ровно один сценарий: read-process-write внутри Kafka. Прочитали из топика A, посчитали, записали в топики B и C, и всё это либо видно потребителям целиком, либо не видно вовсе.

Механика по шагам:

1. Продюсер задаёт transactional.id, стабильный идентификатор, переживающий рестарт (например, billing-worker-p3, привязанный к партиции, а не случайный UUID).
2. initTransactions() находит transaction coordinator и получает PID с новым epoch. Все старые продюсеры с тем же transactional.id и меньшим epoch получают ProducerFenced. Так работает zombie fencing: подвисший старый инстанс не сможет дописать в транзакцию после того, как его заменили.
3. Записи внутри транзакции пишутся в целевые партиции сразу, но помечены как транзакционные.
4. sendOffsetsToTransaction(offsets, consumer.groupMetadata()) кладёт коммит входных offset в ту же транзакцию, и связка «прочитал и записал» становится атомарной.
5. commitTransaction(): координатор пишет решение в __transaction_state и рассылает во все затронутые партиции control batch (маркер COMMIT или ABORT).

На стороне чтения нужен isolation.level=read_committed. Такой консьюмер видит записи только до LSO (last stable offset, offset первой ещё не завершённой транзакции) и отфильтровывает записи прерванных транзакций по маркерам. А по умолчанию стоит read_uncommitted, отсюда классический подвох «включили транзакции, а дубликаты и грязные данные остались»: консьюмера просто забыли переключить.

Библиотеки Go

8.3RabbitMQ и альтернативы

RabbitMQ устроен зеркально Kafka: вместо «лога, который читают указателями» тут умный роутер с очередями, из которых сообщения исчезают после подтверждения. Отсюда гибкая маршрутизация, per-message ack, DLX и TTL, и отсюда же границы: историю не перечитать, одну очередь тяжело масштабировать.

Модель AMQP 0-9-1: кто здесь кто

RabbitMQ построен на открытом протоколе AMQP (Advanced Message Queuing Protocol) версии 0-9-1. Кроме передачи байтов по сети, протокол задаёт модель маршрутизации: как сообщение само находит дорогу к нужной очереди.

Эту модель проще всего понять на обычной почте. Письмо ты не несёшь адресату в квартиру, а сдаёшь в сортировочный центр, написав на конверте адрес. Дальше центр сверяется со своими таблицами и решает, в какие почтовые ящики положить копии. На языке AMQP сортировочный центр называется exchange (обменник), адрес на конверте служит routing key (ключом маршрутизации), а строчка в таблице «письма с таким адресом класть в такой-то ящик» зовётся binding (привязка). Почтовый ящик, где письмо лежит, пока его не заберут, и есть queue (очередь).

Аналогия держится почти до конца, а ломается в одном месте: сортировочный центр ничего не хранит и ничего не возвращает. Если ни одна строчка таблицы не подошла, письмо не уходит обратно отправителю и не ложится в «невостребованные», а просто исчезает. Об этом подробнее ниже.

На собесе первым делом проговори: продюсер в RabbitMQ никогда не публикует в очередь. Он публикует в exchange с каким-то routing key. Буфера в exchange нет, это чистая функция «сообщение + routing key + заголовки → список очередей». Правило, по которому очередь подписана на exchange, называется binding и само содержит binding key (или набор заголовков для headers).

Полный путь сообщения:

• Connection — одно TCP-соединение (и TLS-сессия) до брокера. Дорогое, живёт долго.
• Channel — логический подканал: внутри одного соединения их мультиплексируется много. Все операции (publish, consume, ack, qos, declare) идут через канал. Канал между горутинами не делят: в Go правило «один канал — одна горутина». amqp091-go прикрывает publish и ack мьютексом, но ответ синхронного вызова (declare, qos) на общем канале может достаться чужой горутине, а в других клиентах перемешанные фреймы рвут соединение с 505 UNEXPECTED_FRAME. Ошибка протокола закрывает канал (или всё соединение), и клиент должен уметь его пересоздать.
• Exchange маршрутизирует. Бывает direct, topic, fanout, headers; отдельно стоит системный default exchange с пустым именем "". Он работает как direct, и к нему каждая очередь автоматически привязана по своему имени. Поэтому «наивный» пример publish("", "my-queue", body) и работает без объявления обменника: ты попал в default exchange с routing key, равным имени очереди.
• Queue — единственное место, где сообщение реально лежит. Это FIFO-структура, и в classic-варианте она живёт на одном узле кластера, на том, где её объявили.
• Consumer либо подписывается (basic.consume, push-модель), либо сам опрашивает очередь (basic.get, pull). Второе почти всегда антипаттерн: round-trip на каждое сообщение.

Из этой модели следует неприятная вещь: если сообщение не совпало ни с одной привязкой, оно молча исчезает. Не в DLX и не в «unroutable», а в никуда. Чтобы такие потери увидеть, публикуют с флагом mandatory=true и слушают basic.return (в Go через ch.NotifyReturn(...)) либо вешают на exchange alternate-exchange, куда стекается вся «непонятная почта».

Механика routing key на примерах

Чаще всех работает topic: почти в любом проекте хватает одного topic-обменника на домен. Ключ состоит из слов через точку, это иерархия длиной до 255 байт. Спецсимволов всего два: * заменяет ровно одно слово, # — ноль или больше слов.

Ключ проектируют от общего к частному и кладут в него только то, на что кто-то реально захочет подписаться: домен.сущность.событие.регион. Идентификаторы (order.created.42718) класть в ключ можно, но бессмысленно: на конкретный id всё равно никто не подпишется, а таблица привязок в брокере строится как дерево и растёт.

// Declare идемпотентен, звать его можно при каждом старте.
// Повторный declare с другими параметрами вернёт PRECONDITION_FAILED и закроет канал.
err := ch.ExchangeDeclare(
    "ex.orders", // имя
    "topic",     // тип
    true,        // durable: переживёт рестарт брокера
    false,       // autoDelete: не удалять при отключении последней привязки
    false,       // internal: можно публиковать снаружи
    false,       // noWait: ждём подтверждения от брокера
    nil,
)

q, err := ch.QueueDeclare(
    "q.billing.orders", // именованная очередь: имя = контракт владельца-консьюмера
    true,               // durable
    false,              // autoDelete
    false,              // exclusive: доступна и другим соединениям
    false, nil,
)

// У очереди сколько угодно привязок, и работают они как OR, а не AND.
_ = ch.QueueBind(q.Name, "order.created.#", "ex.orders", false, nil)
_ = ch.QueueBind(q.Name, "order.paid.#",    "ex.orders", false, nil)

Подтверждения на стороне консьюмера: ack, nack, reject

Когда брокер отдаёт сообщение консьюмеру, оно не удаляется из очереди, а переходит в состояние unacked (доставлено, но не подтверждено) и остаётся привязанным к каналу. Каждая доставка получает delivery tag, монотонный счётчик внутри канала. Это не идентификатор сообщения и не сквозной номер: после реконнекта нумерация начинается заново. Закрыть доставку консьюмер может тремя способами:

Если канал или соединение закрывается (падение пода, сетевой обрыв, ошибка протокола), все unacked-сообщения этого канала автоматически возвращаются в очередь и будут доставлены заново с флагом redelivered=true. Так в RabbitMQ и устроен at-least-once: таймаут видимости, как в SQS, не нужен, гарантию даёт сам разрыв TCP-соединения.

Prefetch (basic.qos): сколько сообщений можно держать в руках

Prefetch («предвыборка») задаёт, сколько сообщений брокер может выдать одному консьюмеру авансом, не дожидаясь подтверждений по уже выданным. Это обратное давление из главы 8.1, сжатое в одно число: официант несёт не больше трёх тарелок, и четвёртую ему не дадут, пока он не поставит хотя бы одну.

basic.qos(prefetchSize, prefetchCount, global) задаёт максимум неподтверждённых доставок. Когда их набралось prefetchCount, этот потребитель ничего нового не получит: следующее сообщение уйдёт кому-то другому. prefetchSize (лимит в байтах) на практике не используют. global=false означает «лимит на каждого консьюмера канала», global=true — «лимит на весь канал суммарно».

По умолчанию prefetch не ограничен, и хуже дефолта не придумать: первый подключившийся консьюмер выгребает всю очередь себе в память, остальные простаивают.

Как подобрать значение? Каждое подтверждение стоит сетевого round-trip, поэтому prefetch=1 даёт идеально ровное распределение и низкий throughput: консьюмер простаивает, пока ack идёт туда, а следующее сообщение обратно. Ориентиры такие:

Requeue и бесконечный цикл переобработки

nack(requeue=true) возвращает сообщение в очередь. Вопрос в том, куда: RabbitMQ старается поставить его как можно ближе к исходной позиции, то есть практически в голову. Если консьюмер один, а ошибка постоянная (невалидный JSON, отсутствующая запись, битая схема), выходит цикл: доставка → ошибка → requeue → доставка через микросекунду. Ретраем такое не назовёшь: получается горячий цикл на 100 % CPU, который заодно держит за собой всю очередь.

// Плохо: вечный цикл на постоянной ошибке
msgs, _ := ch.Consume(q, "", false, false, false, false, nil)
for d := range msgs {
    if err := handle(d.Body); err != nil {
        d.Nack(false, true) // requeue = true
        continue            // и через мгновение снова сюда
    }
    d.Ack(false)
}

// Хорошо: разделяем транзиентные и постоянные ошибки
for d := range msgs {
    err := handle(d.Body)
    switch {
    case err == nil:
        d.Ack(false)
    case errors.Is(err, ErrPermanent), attempts(d) >= maxAttempts:
        // DLX у очереди один, и он ведёт в retry. В DLQ кладём сами.
        if publishToDLQ(ctx, ch, d) == nil {
            d.Ack(false)
        } // не вышло — не подтверждаем, сообщение вернётся
    default:
        d.Nack(false, false) // через DLX в retry-очередь с TTL, см. ниже
    }
}

Ещё одна неприятность: в AMQP нет встроенного счётчика попыток. Булево поле redelivered скажет «это не первая доставка», но номер попытки не назовёт. Считать попытки можно тремя способами:

• Через x-death: при каждом проходе через dead-lettering брокер сам наращивает count в заголовке x-death. Самый честный вариант, руками ничего писать не надо, но работает он, только если ретраи идут через DLX.
• Своим заголовком: консьюмер читает x-retry-count, увеличивает, публикует сообщение заново и подтверждает исходное. Работает всегда, но публикация + ack не атомарны: упал между ними и потерял сообщение (или продублировал, если порядок обратный).
• Внешним счётчиком в Redis по message_id. Он нужен, когда попытки считают сквозь несколько разных очередей и сервисов.

DLX — dead letter exchange

Термин dead letter пришёл с почты: так называют «мёртвое письмо», которое не удалось ни доставить, ни вернуть отправителю. Такие письма уходят в отдельный отдел, где их разбирают вручную. Брокеры взяли и слово, и идею: необработанное сообщение не выбрасывают, а откладывают в сторону, чтобы человек потом посмотрел, что с ним не так. Отсюда и DLQ (dead letter queue), «очередь битых сообщений».

DLX (dead letter exchange) не «специальный тип очереди», а обычный exchange: в него очередь пересылает сообщения, которые из неё выбыли не по-хорошему. Задают его двумя аргументами при объявлении очереди, а в проде правильнее через policy, чтобы не передекларировать очередь ради смены маршрута:

args := amqp.Table{
    "x-dead-letter-exchange":    "dlx.orders",   // куда переслать
    "x-dead-letter-routing-key": "orders.dead",  // с каким ключом; без него сохранится исходный
}
ch.QueueDeclare("q.orders", true, false, false, false, args)

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

При каждом прохождении брокер дописывает запись в заголовок x-death. Там лежит массив структур с полями queue, reason, count, exchange, routing-keys, time. Другого штатного счётчика попыток в AMQP нет: x-death[0].count честно говорит, сколько раз сообщение уже возвращалось через dead-lettering.

Отложенный ретрай: лестница из TTL + DLX

В RabbitMQ нет встроенной «доставки через 30 секунд». Стандартный трюк: завести очередь без консьюмеров, у которой есть x-message-ttl и DLX обратно на рабочий обменник. Сообщение полежит там положенное время, «протухнет» и вернётся в работу.

Durable, persistent, confirms: четыре независимых условия

«Сообщение переживёт рестарт брокера» требует не одной галочки, а конъюнкции четырёх условий, и каждое отвечает за свой кусок пути. Провалить достаточно одно.

Комбинации ведут себя ровно по таблице, без сюрпризов: transient-сообщение в durable-очереди пропадёт при рестарте (при нехватке памяти его могут выгрузить на диск, но это экономия памяти, а не долговечность), persistent-сообщение в non-durable-очереди пропадёт вместе с очередью.

// Полный «надёжный» publish
if err := ch.Confirm(false); err != nil { return err }          // включаем confirms на канале
confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 256)) // окно на 256 неподтверждённых
returns  := ch.NotifyReturn(make(chan amqp.Return, 8))          // сюда придут unroutable

err := ch.PublishWithContext(ctx, "x.orders", "order.created",
    true,  // mandatory: не смаршрутизировалось — верни, а не выбрось
    false, // immediate: RabbitMQ не поддерживает с 3.0, всегда false
    amqp.Publishing{
        DeliveryMode: amqp.Persistent, // = 2, иначе тело только в RAM
        MessageId:    evt.ID,          // ключ дедупликации для консьюмера
        Timestamp:    time.Now(),
        ContentType:  "application/json",
        Body:         body,
    })

c := <-confirms
if !c.Ack {
    // брокер не принял: публикуем заново, дубликаты тут норма
}

Classic, quorum, streams

• Classic queue, исходный тип, живёт целиком на одном узле. Реплицировать её умело только устаревшее зеркалирование (ha-mode): его объявили deprecated и выпилили в 4.0. Она дешёвая по ресурсам и годится для некритичных задач и очередей, которые не жалко потерять.
• Quorum queue — реплицированная очередь на Raft, современный дефолт для всего, что нельзя терять. Всегда durable, подтверждение = запись на большинство узлов, есть штатная обработка poison message через x-delivery-limit. Платить приходится диском и памятью (Raft-лог держится до snapshot), отсутствием transient-режима, exclusive-очередей и global-prefetch.
• Stream (3.9+) — append-only лог с офсетами: несколько независимых читателей, перечитывание с любой позиции, ретеншен по времени и размеру. По сути так RabbitMQ отвечает Kafka, когда нужно не «раздать задачи», а «многие читают одну ленту».

Kafka vs RabbitMQ: где проходит граница

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

Как формулировать выбор

• Kafka, если поток служит лентой фактов, которую читают несколько независимых систем и хотят перечитать: событийная интеграция между доменами, аналитика и стриминг, CDC из БД, аудит, материализация проекций, метрики. Признак: «нужен replay» или «потребителей будет больше, чем сейчас».
• RabbitMQ, если по потоку идут задачи: «сконвертируй видео», «отправь письмо», «сходи в 1С». Признаки: нужна сложная маршрутизация по атрибутам, per-message ретраи с задержкой, приоритеты, RPC-стиль request/reply, много мелких разнородных очередей, команда не хочет содержать Kafka.
• Оба сразу тоже нормальная архитектура, никакого хаоса в этом нет: события домена в Kafka, фоновые задачи в RabbitMQ. Ответ «взяли бы Kafka, потому что она модная» идёт в минус, ответ «зависит от того, лента это или задачи» в плюс.

NATS и JetStream

Core NATS — предельно простой брокер на Go: один статический бинарник, десятки мегабайт памяти, задержки в десятки микросекунд. Модель называется subject-based messaging: публикуют в тему вида orders.eu.created, подписываются по шаблону с * (ровно одно слово) и > (хвост целиком). И в отличие от остальных, Core NATS не хранит ничего. Если в момент публикации подписчика нет, сообщение просто исчезает: at-most-once, никаких ack. Зато из коробки есть то, чего нет у соседей: queue groups (подписчики с одинаковым именем группы делят поток, получаются competing consumers без объявления очередей) и встроенный request/reply с автоматическим inbox-subject для ответа.

JetStream добавляет поверх слой персистентности и включается флагом. С ним появляются:

• Stream — лог, захватывающий один или несколько subjects, с политикой хранения: limits (как Kafka: держим, пока влезает по max_msgs, max_bytes, max_age), interest (удаляем, когда все подписчики подтвердили) и workqueue (как очередь: сообщение удаляется, как только его подтвердил один консьюмер). Хранить можно в file или memory, реплицирует Raft (replicas: 3).
• Consumer — именованная позиция в стриме, durable или ephemeral, push или pull, с AckPolicy (none/all/explicit), AckWait, MaxDeliver, BackOff (готовая лестница задержек ретрая, которую в RabbitMQ приходится собирать из очередей) и FilterSubject. MaxAckPending работает как prefetch.
• Дедупликация из коробки: заголовок Nats-Msg-Id и duplicate_window на стриме. Брокер сам отбросит повторную публикацию с тем же идентификатором внутри окна. В Kafka ради этого включают идемпотентного продюсера, в RabbitMQ пишут руками.
• KV и Object store надстроены над стримом: key-value с watch на изменения и хранилище блобов. Часто они закрывают потребность, ради которой иначе тянули бы ещё и Redis.

rabbitmqctl set_policy dlx-all "^q\\.orders\\." \
  '{"dead-letter-exchange":"dlx.orders","message-ttl":600000,"max-length":100000}' \
  --apply-to queues

8.4Паттерны асинхронной интеграции

Паттерны этой главы лечат одну и ту же беду: у нас две системы, а транзакция только в одной. База данных умеет атомарность, брокер умеет доставку, общего коммита между ними нет. Outbox переносит публикацию внутрь транзакции БД, inbox переносит дедупликацию внутрь транзакции консьюмера. Остальное касается того, что класть в сообщение и что делать, когда сообщений становится больше, чем сил их переварить.

Двойная запись — корень всех бед

Задача выглядит безобидно: сохранить заказ в PostgreSQL и опубликовать событие OrderCreated, чтобы отреагировали склад, биллинг и аналитика. Два действия, две разные системы, и ни одного способа сделать их атомарно. Какой порядок ни выбери, между ними остаётся окно, в котором процесс может умереть — под убьёт OOM-killer, нода уедет на обслуживание, сеть моргнёт.

Transactional outbox

Outbox буквально значит «ящик исходящих», как в почтовом клиенте: письмо ты уже написал и нажал «отправить», но оно ещё не ушло. Зато оно точно существует и точно уйдёт — просто позже. Relay («релей», передатчик) — фоновый процесс, который разгребает этот ящик: берёт неотправленное и толкает в брокер. Вся конструкция называется transactional outbox — «транзакционный ящик исходящих»: запись в него идёт внутри той же транзакции БД, что и само бизнес-изменение.

Выход в том, чтобы сделать публикацию частью той же транзакции БД. Раз атомарности между базой и брокером не бывает, событие записывают в ту же базу, в таблицу outbox, тем же COMMIT. Дальше релей вычитывает эту таблицу и публикует в брокер. Атомарность обеспечивает СУБД, доставку берёт на себя релей, и окна между ними нет: закоммитилось либо всё, либо ничего.

CREATE TABLE outbox (
    id             bigserial PRIMARY KEY,
    aggregate_type text        NOT NULL,          -- 'order'  — отсюда имя топика
    aggregate_id   text        NOT NULL,          -- '42'     — станет ключом партиционирования
    event_type     text        NOT NULL,          -- 'OrderCreated'
    payload        jsonb       NOT NULL,          -- тело события
    headers        jsonb       NOT NULL DEFAULT '{}'::jsonb,  -- traceparent, schema_version
    created_at     timestamptz NOT NULL DEFAULT now(),
    published_at   timestamptz                    -- NULL = ещё не отправлено
);

-- частичный индекс маленький: в нём только неотправленные строки
CREATE INDEX outbox_unpublished_idx ON outbox (id) WHERE published_at IS NULL;

func (s *Service) CreateOrder(ctx context.Context, o Order) error {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil { return err }
    defer tx.Rollback() //nolint:errcheck

    if _, err = tx.ExecContext(ctx,
        `INSERT INTO orders (id, user_id, total, status) VALUES ($1,$2,$3,'new')`,
        o.ID, o.UserID, o.Total); err != nil {
        return err
    }

    payload, _ := json.Marshal(OrderCreated{OrderID: o.ID, UserID: o.UserID, Total: o.Total})
    if _, err = tx.ExecContext(ctx,
        `INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload, headers)
         VALUES ('order', $1, 'OrderCreated', $2, $3)`,
        o.ID, payload, traceHeaders(ctx)); err != nil {
        return err
    }

    return tx.Commit() // ← всё становится реальным только здесь
}

Relay через polling

Самый простой вариант: фоновая горутина или отдельный воркер раз в N миллисекунд забирает пачку неотправленных строк, публикует их и помечает. Нетривиальная деталь тут одна — FOR UPDATE SKIP LOCKED: с ней можно держать несколько экземпляров релея, и они не блокируют друг друга и не берут одни и те же строки.

const batch = `
  SELECT id, aggregate_type, aggregate_id, event_type, payload, headers
    FROM outbox
   WHERE published_at IS NULL
   ORDER BY id
   LIMIT 100
     FOR UPDATE SKIP LOCKED`      // ← конкурентные релеи разбирают разные строки

func (r *Relay) tick(ctx context.Context) error {
    tx, err := r.db.BeginTx(ctx, nil)
    if err != nil { return err }
    defer tx.Rollback() //nolint:errcheck

    rows, err := tx.QueryContext(ctx, batch)
    if err != nil { return err }
    msgs, ids := scanBatch(rows)

    for i, m := range msgs {
        // ключ = aggregate_id: события одного заказа лягут в одну партицию;
        // порядок между ними держится, пока релей один
        if err := r.producer.Publish(ctx, topicFor(m), m.AggregateID, m.Payload, m.Headers); err != nil {
            return err            // строки останутся неотмеченными, отправим в следующий тик
        }
        _ = i
    }

    _, err = tx.ExecContext(ctx,
        `UPDATE outbox SET published_at = now() WHERE id = ANY($1)`, pq.Array(ids))
    if err != nil { return err }
    return tx.Commit()
}

Что ещё надо предусмотреть в polling-релее:

• Уборка. Таблица растёт со скоростью потока событий. Чистят либо пачками через DELETE ... WHERE published_at < now() - interval '3 days' (не одним запросом на миллион строк — иначе выйдет долгая транзакция и раздутый WAL), либо DELETE сразу после публикации, либо секционированием по дню: старую секцию отцепляют (DETACH PARTITION) и удаляют через DROP TABLE. Без уборки деградируют индекс и autovacuum, а за ними и весь сервис, а не один релей.
• Интервал опроса — компромисс между задержкой и нагрузкой. При 100 мс это 10 лишних запросов в секунду на пустой таблице (дёшево благодаря частичному индексу), при 5 с событие ждёт 5 с. Разумно взять короткий интервал плюс сигнал NOTIFY outbox после коммита, чтобы релей просыпался сразу.
• Порядок. Один релей с ORDER BY id публикует по порядку, но у нескольких экземпляров со SKIP LOCKED порядка нет даже внутри агрегата: два релея могут взять события 4 и 5 одного заказа в разные пачки, и пятое уйдёт раньше. Ключ aggregate_id положит их в одну партицию, но в том порядке, в каком они пришли. Если порядок внутри агрегата важен, берут схему «один релей, лидер через advisory lock» или CDC.
• Идемпотентность продюсера. С enable.idempotence уйдут дубликаты от ретраев внутри одной сессии продюсера, но не те, что появятся при перезапуске релея.

Relay через CDC (Debezium)

При CDC (Change Data Capture, «захват изменений данных») изменения в базе вычитывают не запросами к таблицам, а из журнала транзакций самой СУБД (WAL в PostgreSQL, binlog в MySQL). База и так пишет в этот журнал каждое изменение, чтобы восстановиться после сбоя. CDC-инструмент подключается к нему как ещё одна реплика и превращает каждую запись журнала в сообщение, не делая ни одного запроса к рабочей базе.

Второй вариант релея не спрашивает базу, а читает её журнал. Debezium подключается к PostgreSQL через логический слот репликации (к MySQL как реплика по binlog) и превращает каждое изменение строки в сообщение Kafka. Если направить его на таблицу outbox, получится релей без единого SELECT в рабочую базу.

Inbox и дедупликация на приёме

Inbox («ящик входящих») — зеркальная пара к outbox. В outbox лежит то, что мы обязаны отправить, а в inbox — то, что мы уже приняли и обработали. Магии в нём нет: обычная таблица с ключами сообщений, по которой консьюмер проверяет, не видел ли он это сообщение раньше.

У консьюмера та же история, только зеркальная: надо сделать работу и запомнить, что сообщение обработано, — снова две операции. Если запоминать отдельным запросом после коммита работы, между ними опять появляется окно: упали, потеряли отметку, обработали второй раз. Решение симметрично outbox: отметку кладут в ту же транзакцию, что и бизнес-работу.

CREATE TABLE processed_messages (
    message_id  text        NOT NULL,
    consumer    text        NOT NULL,      -- имя consumer group: каждый обрабатывает своё
    processed_at timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (consumer, message_id)     -- уникальность и есть весь механизм
);
-- Секционировать по processed_at нельзя: ключ обязан включать processed_at,
-- и дедупликация сломается. Старое удаляют пачками по индексу.
CREATE INDEX ON processed_messages (processed_at);

func (c *Consumer) Handle(ctx context.Context, msg Message) error {
    tx, err := c.db.BeginTx(ctx, nil)
    if err != nil { return err }
    defer tx.Rollback() //nolint:errcheck

    // 1) сначала отметка: INSERT блокирует ключ и сериализует
    //    конкурентные доставки одного сообщения
    res, err := tx.ExecContext(ctx,
        `INSERT INTO processed_messages (consumer, message_id) VALUES ($1, $2)
         ON CONFLICT DO NOTHING`, c.name, msg.ID)
    if err != nil { return err }
    if n, _ := res.RowsAffected(); n == 0 {
        return tx.Commit() // дубликат: уже обрабатывали, просто подтверждаем брокеру
    }

    // 2) бизнес-работа в той же транзакции
    if err := c.apply(ctx, tx, msg); err != nil {
        return err // rollback снимет и отметку: сообщение вернётся, обработаем заново
    }

    return tx.Commit() // 3) отметка и результат становятся видимыми вместе
}

Детали, на которых валятся

• Ключ дедупликации. Лучше всего подходит message_id, который продюсер сгенерировал один раз и переиспользует при всех ретраях (в outbox это естественно: id строки). Годится и бизнес-ключ вида order_id + event_type + version. Хеш тела не годится: ретрай с обновлённым timestamp даст другой хеш и проедет как новое сообщение, а два легитимных одинаковых события (два одинаковых списания по 100 ₽), наоборот, схлопнутся в одно.
• Область уникальности задаёт пара «консьюмер + message_id», а не один message_id. Иначе первая же группа «съест» сообщение у всех остальных.
• Порядок внутри транзакции: сначала INSERT отметки, потом работа. Так конкурентные доставки одного сообщения (две реплики консьюмера после ребалансировки) упрутся в блокировку по первичному ключу и сериализуются, а не выполнят работу дважды.
• Ack брокеру уходит строго после COMMIT. Ack до коммита возвращает нас к at-most-once со всеми его потерями.
• Ретеншен. Окно дедупликации должно быть шире самого долгого возможного повтора: retention топика, длительности retry-лестницы, планового replay. Чистят пачками DELETE по индексу processed_at, а не одним запросом на миллион строк. Если чистить слишком агрессивно, повторное проигрывание старого топика создаст полный комплект дубликатов.
• Побочные эффекты вне БД. Отправленное письмо транзакция не откатит, вызов внешнего API не отменит. Помогает идемпотентный внешний вызов с Idempotency-Key или запись «надо отправить письмо» в свой outbox — тогда эффект уйдёт наружу из релея, а не из середины транзакции.

Competing consumers

Несколько одинаковых консьюмеров разбирают одну очередь. Так сервис масштабируют добавлением подов: брокер сам следит, чтобы каждое сообщение досталось ровно одному обработчику, и приложению не нужны ни шардирование, ни координация.

Что ломается при переходе от одного консьюмера к нескольким:

• Порядок. Это главное. Два консьюмера работают как два независимых потока: сообщение №2 может завершиться раньше №1, и «заказ оплачен» обработается раньше «заказ создан». Если порядок важен, его сохраняют шардированием по ключу: в Kafka партиционируют по order_id, в RabbitMQ берут плагин consistent-hash exchange или явный набор очередей на шард. Порядок тогда гарантирован внутри ключа, а бизнесу большего и не надо: события разных заказов независимы.
• Гонки на общих данных. Параллельные обработчики одного агрегата дерутся за строку в БД. Спасает шардирование (см. выше), оптимистическая блокировка по версии или SELECT ... FOR UPDATE, но последнее берут осознанно: оно сериализует обработку и ставит потолок пропускной способности.
• Poison message. Одно вечно падающее сообщение при requeue=true съедает всех консьюмеров по очереди. Ограничение попыток и DLQ обязательны.
• Долгие задачи. Пока обработчик занят, брокер ждёт признаков жизни: в Kafka это max.poll.interval.ms (превысишь — консьюмера исключат из группы и запустят ребалансировку), в JetStream AckWait, в SQS продление visibility timeout. Долгие задачи лучше дробить или уносить в отдельный поток с большими таймаутами, чтобы не тащить эти настройки на весь сервис.

Fanout / multicast и job distribution

Эти два режима постоянно путают, хотя разница ровно в одном: сколько раз сообщение будет обработано. При fanout каждый подписчик получает свою копию (широковещание). При job distribution оно достаётся ровно одному из группы (конкуренция). Оба режима даёт одна и та же инфраструктура, и ошибка в конфигурации молча превращает один в другой.

Отдельный вопрос — кто плодит копии. В Kafka копия одна: N групп читают один лог, данные хранятся однократно. RabbitMQ при fanout физически размножает сообщение по очередям: у 10 подписчиков будет 10 копий в памяти и на диске. С крупными сообщениями это быстро становится заметно, и лечится claim check: в сообщение кладут ссылку на объект в S3, а не сам объект.

Что класть в событие: notification, state transfer, event sourcing

Наполнить сообщение можно тремя способами, и каждый по-своему распределяет связанность между сервисами. Этот выбор важнее выбора брокера: от него зависит, сможет ли потребитель работать, когда источник лежит.

Схемы сообщений и их эволюция

Сообщение — такой же публичный API, только без компилятора и без единого места, где его можно посмотреть. HTTP-контракт хотя бы ломается сразу и в тестах, а событие ломается через полгода у команды, о существовании которой ты не знал, — и не падением, а тихим неправильным поведением. Отсюда две практики: описывать схему явно и проверять совместимость машиной.

Protobuf

В protobuf контракт держится на номерах полей, а не на именах: имена в wire-формате не передаются вовсе. Из этого и следуют все правила.

message OrderCreated {
  string order_id = 1;
  string user_id  = 2;
  int64  total    = 3;         // в копейках: менять тип потом будет нельзя

  reserved 4, 7 to 9;          // номера удалённых полей больше не занимать
  reserved "discount_code";    // и имя тоже, чтобы не вернулось с другим смыслом

  repeated Item items    = 5;
  optional string coupon = 6;  // explicit presence: отличаем "нет значения" от ""
}

Avro и Schema Registry

Avro устроен иначе: схема не зашита в код, а сопровождает данные. Читатель делает schema resolution: сопоставляет writer schema и reader schema по именам полей и подставляет default для тех, которых нет. Отсюда правило: у любого добавляемого поля должен быть default, иначе старые данные перестанут читаться.

Schema Registry («реестр схем») — отдельный маленький сервис рядом с брокером: он хранит все версии схем сообщений и раздаёт им номера. Нужен он потому, что сама Kafka о содержимом сообщений ничего не знает и проверить его не может: для неё это просто байты. В реестре договорённость о формате становится машинно проверяемой.

Confluent Schema Registry раскладывает версии по subject (обычно <topic>-value) и выдаёт им числовые id. В сообщение пишется префикс: magic byte 0 + 4 байта schema id, дальше тело. Консьюмер вытаскивает id, тянет схему из реестра (и кэширует) и декодирует. Приятный побочный эффект, который стоит упомянуть: схема не передаётся с каждым сообщением — только 5 байт.

Но главное в реестре не хранение, а проверка совместимости при регистрации: несовместимую схему он просто не примет, и продюсер упадёт на старте, а не на проде через месяц.

В Kafka по умолчанию стоит BACKWARD, потому что в топике лежат старые записи (retention неделя, а то и вечность при compaction), и новый консьюмер обязан их прочитать. FORWARD берут, когда консьюмеров обновить нельзя: внешние клиенты, мобильные приложения, партнёрские интеграции. Продюсер уходит вперёд, а старый код должен продолжать работать. FULL нужен, когда порядок выкатки не контролируется вовсе, а в микросервисах это скорее норма, чем исключение.

Backpressure: консьюмер не успевает

Напомним определение из главы 8.1: при backpressure (обратном давлении) медленный получатель заставляет быстрого отправителя сбавить темп. Здесь разберём, что делать, если этого механизма не хватило и невыполненная работа начала копиться.

Сначала диагноз, потом лечение. Растущий лаг сам по себе ничего не значит: при пике он растёт и возвращается, при проблеме растёт монотонно. Различают их по одной величине: скорость разгребания = consume_rate − produce_rate. Если она устойчиво отрицательная, никакой буфер не поможет — вопрос лишь в том, когда он кончится. На собесе полезно произнести такую оценку: время до катастрофы = свободный объём буфера / (produce − consume). Она сразу переводит разговор из «надо что-то делать» в «у нас четыре часа».

Инструменты по порядку применения

• Горизонтальное масштабирование консьюмеров. Его пробуют первым и в него же первым упираются: в Kafka потолок задаёт число партиций, лишние поды в группе просто простаивают. В RabbitMQ потолка почти нет, но растёт нагрузка на сам процесс очереди. Прежде чем масштабировать, стоит проверить, что консьюмеры заняты работой, а не ждут БД: если ждут, новые поды только добьют базу.
• Ускорить обработку. Обычно тут и лежит основной резерв: батчить записи в БД вместо построчных, убрать N+1, вынести побочные вызовы, распараллелить внутри одного консьюмера воркер-пулом по hash(key) % N, коммитя только сплошной префикс обработанных офсетов. Так параллелизм растёт без потери порядка по ключу.
• Увеличить число партиций, понимая цену. (а) Меняется hash(key) % N, ключ уезжает в другую партицию, и порядок по ключу ломается на границе изменения: старые события заказа лежат в старой партиции, новые в новой, и два консьюмера обрабатывают их одновременно. (б) Уменьшить число партиций нельзя вообще. (в) Растут файловые дескрипторы, память на брокерах, время ребалансировки и выборов лидеров. (г) Уже лежащие сообщения не перераспределяются, так что сразу легче не станет.
• Throttling продюсера, то есть ограничение на источнике: quotas в Kafka (producer_byte_rate на client.id/user: брокер задерживает ответы, и продюсер тормозит сам), rate limiter в сервисе, 429 на HTTP-границе. Обратное давление должно доходить до того, кто порождает трафик, иначе оно копится где-то посередине и рвётся там.
• Отбрасывание и приоритеты (load shedding). Заранее решить, что можно потерять. Аналитику, телеметрию, «пользователь открыл экран» можно, платежи нельзя. Технически это разные топики или очереди под разные классы с отдельными пулами консьюмеров (bulkhead); в RabbitMQ x-max-length с overflow=drop-head (вытесняем самое старое) или reject-publish (отказываем на входе, что честнее); TTL для данных, которые через минуту бессмысленны, — котировки, статусы онлайна. Осознанное отбрасывание всегда лучше неосознанного: второе всё равно случится, только жертву выберет само.
• Буферизация и её предел. Брокер и есть буфер, и копить правильно именно в нём: диск дешевле памяти консьюмера. Но он конечен: его ограничивают retention.ms/retention.bytes, x-max-length и размер диска. Буфер даёт время дожить до пятого пункта, но не заменяет его.