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

Начни с того, кто инициирует передачу, где хранится сообщение и что означает подтверждение. Разбери at-most-once, at-least-once и идемпотентность на временной линии, а затем оцени, где порядок вообще существует.

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

При синхронном вызове мало дождаться ответа: вызываемый сервис обязан быть жив прямо сейчас. Такая связанность называется 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-теорема: в асинхронной системе, где может отказать даже один узел, детерминированный консенсус за конечное время невозможен.

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

Операция идемпотентна, если даёт один и тот же результат, сколько бы раз её ни повторили: один раз, три раза, сто. Так устроена кнопка вызова лифта. Нажал её один раз или продолбил пятнадцать - лифт приедет один раз, состояние «лифт вызван» от повторов не меняется. А вот кнопка «взять талончик» в очереди неидемпотентна: каждое нажатие выдаёт новый номер. Слово латинское: 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()
}

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

Персистентность гарантирует, что сообщение переживёт перезапуск брокера. Для этого сообщение мало принять: его записывают туда, откуда оно достанется после включения. Обратный режим называется 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 (подробно в 8.2) не доступна, партиция становится недоступной, но не расходящейся. Это осознанный выбор CP в терминах CAP.
• RabbitMQ. Классические зеркалируемые очереди 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);

Что стоит запомнить

• At-least-once допускает повторную обработку, поэтому обработчик должен быть идемпотентным.
• Ack означает завершение этапа доставки, но его точный смысл зависит от брокера и момента фиксации.
• Порядок обычно ограничен очередью, partition или ключом, а не всем кластером.

Частые ошибки

• Не отправляй ack до durable business commit: сбой после ack теряет работу.
• Не обещай exactly-once между брокером и внешней БД без общей транзакции или дедупликации.
• Дедупликационный ключ должен представлять логическую операцию и быть уникален в scope consumer.

delivery M -> INSERT order -> crash -> redelivery M

if err := d.Ack(false); err != nil { return err }
return repo.InsertOrder(ctx, msg.Order)