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

Разбери двойную запись между БД и брокером и собери outbox/inbox вокруг локальных транзакций. Затем проверь, что сообщение достаточно версионировано, конкурентно опубликовано без потери и обработано при at-least-once доставке.

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

Задача выглядит безобидно: сохранить заказ в 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 изменения в базе вычитывают не запросами к таблицам, а из журнала транзакций самой СУБД (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: консьюмер не успевает

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

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

• Горизонтальное масштабирование консьюмеров. Его пробуют первым и в него же первым упираются: в Kafka потолок consumer group задаёт число партиций, лишние поды в группе просто простаивают. В 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 и размер диска. Буфер даёт время дожить до пятого пункта, но не заменяет его.

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

• Outbox атомарно фиксирует бизнес-изменение и событие в одной БД-транзакции; relay публикует позже.
• SKIP LOCKED позволяет нескольким relay workers забирать разные строки, но не устраняет повтор после публикации и до отметки published.
• Inbox/processed_messages фиксирует дедупликацию и бизнес-эффект в одной транзакции consumer.

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

• Не помечай outbox опубликованным до broker confirm; иначе сбой теряет событие.
• Если confirm получен, но DB update published_at не зафиксировался, relay опубликует повтор - consumer должен дедуплицировать.
• Не переиспользуй номера и смысл удалённых полей в protobuf/schema evolution.

if err := broker.Publish(ctx, row.EventID, row.Payload); err != nil { return err }
return markPublished(ctx, row.ID)

markPublished(ctx, row.ID)
if err := broker.Publish(ctx, row.Payload); err != nil { return err }

if err := broker.Publish(ctx, row.EventID, row.Payload); err != nil { return err }
return markPublished(ctx, row.ID)