Событийная интеграция: как сервисы общаются без прямых вызовов

Прямой ответ

Событийная интеграция заменяет прямые вызовы публикацией фактов. Сервис не говорит другому «сделай это», а сообщает «это произошло»; заинтересованные сервисы реагируют по-своему.

Различие принципиально:

Событие снижает связанность: новый потребитель добавляется без изменения издателя. Команда нужна там, где требуется конкретное действие от конкретного сервиса.

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

def create_order(session, order: Order) -> None:
    session.add(order)
    session.add(OutboxEvent(                     # та же транзакция
        aggregate_id=order.id,
        event_type="order.created",
        payload=json.dumps(order.to_event()),
    ))
    session.commit()

Отдельный процесс читает таблицу outbox_event и публикует записи в брокер. Поскольку запись события и запись данных происходят в одной транзакции, событие не теряется. О паттерне подробнее — в статье про saga и outbox.

Вторая проблема — доставка. Гарантии брокера дают at-least-once, поэтому потребитель обязан быть идемпотентным: повторная доставка того же события не должна приводить к повторному действию. Практический приём — хранить идентификаторы обработанных событий:

def handle(event: dict) -> None:
    if already_processed(event["event_id"]):
        return
    apply_business_logic(event)
    mark_processed(event["event_id"])

О принципе — в статье про идемпотентность.

Практика: надёжный обмен событиями

1. Пишите событие в прошедшем времени и в терминах домена. order.created описывает факт; createOrder — это команда. Имя события становится частью контракта и живёт дольше реализации.

2. Версионируйте схему. Потребители обновляются в разное время, поэтому добавление поля безопасно, а удаление и переименование — нет. Практика: order.created.v1, order.created.v2 с переходным периодом; о совместимости — в статье про версии API.

3. Не кладите в событие большие данные. Событие содержит идентификаторы и факт, а не полный образ сущности. Потребитель запрашивает детали у владельца, если они нужны.

4. Разделяйте очереди по потребителям. Один топик на всех проще, но замедляет медленного потребителя и не позволяет обработать сбой изолированно.

5. Настраивайте мёртвую очередь. Событие, которое не удаётся обработать после нескольких попыток, уходит в отдельную очередь для разбора, а не блокирует поток; о гарантиях доставки — в статье про надёжность очередей.

6. Учитывайте порядок. Порядок гарантирован только внутри партиции по ключу. Если важен порядок событий одного заказа, ключ партиционирования — идентификатор заказа.

7. Измеряйте отставание. Лаг потребителя показывает, насколько события отстают от фактов. Рост лага — ранний сигнал о проблеме.

Рекомендации:

1. Начинайте с событий для чтения, а не для записи. Обновление локальных копий по событиям проще и безопаснее, чем распределённые транзакции.
2. Фиксируйте схему в репозитории. Схема события — контракт; её изменение должно проходить ревью; о реестре схем — в статье про Schema Registry.
3. Логируйте идентификатор события сквозным образом. Без него невозможно проследить обработку через несколько сервисов.
4. Тестируйте повторную доставку. Тест, который подаёт одно событие дважды, выявляет неидемпотентность до продакшена.
5. Не заменяйте событиями все вызовы. Там, где нужен немедленный ответ, синхронный запрос проще и предсказуемее.

Практический сценарий: событие дублировалось и начисляло бонусы дважды

Контекст: после публикации события order.paid сервис лояльности начислял бонусы. Периодически пользователи получали двойное начисление.

1. Диагностика. В логах потребителя одно и то же событие встречалось дважды с разными смещениями в топике. Идентификатор события совпадал.
2. Причина. Доставка была at-least-once: при перебалансировке потребителей часть сообщений доставлялась повторно. Обработчик не проверял, обработано ли событие ранее, поэтому начисление выполнялось снова.
3. Дополнительная находка. Часть дублей возникала на стороне издателя: задача публикации читала outbox и отправляла записи, не помечая их как опубликованные при сбое отправки.
4. Действия. (а) Добавили таблицу обработанных идентификаторов событий с уникальным ограничением; (б) начисление и отметку об обработке объединили в одну транзакцию; (в) в издателе пометку о публикации перенесли в транзакцию с чтением; (г) добавили тест, подающий одно событие дважды и ожидающий однократного начисления; (д) настроили мёртвую очередь для событий с ошибкой разбора, чтобы они не блокировали поток.
5. Результат. Двойные начисления прекратились; повторная доставка больше не влияет на результат, а тест фиксирует требование идемпотентности.

Вывод: at-least-once — это не исключение, а нормальный режим работы брокера. Идемпотентность потребителя обязательна по умолчанию, а не добавляется после первого инцидента.

FAQ

Чем событийная интеграция отличается от очереди задач? В очереди задач сообщение адресовано конкретному исполнителю и требует действия. Событие описывает факт и может не иметь потребителей вовсе.

Можно ли обойтись без брокера? Для двух сервисов и низкой нагрузки — да, через HTTP-уведомления. Но при этом теряются буферизация и независимость от доступности потребителя.

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

Нужно ли хранить все события? Хранение истории событий полезно для аудита и восстановления, но это отдельное решение — event sourcing; о подходе в статье про event sourcing.

Источники: microservices.io (event-driven, transactional outbox), Kafka docs — ссылки в шапке.