Когда приложение разбивают на несколько сервисов, исчезает главное удобство монолита — единая транзакция базы данных. Раньше один блок with session.begin() гарантировал: либо заказ создан, деньги списаны и склад зарезервирован — либо ничего из этого. В микросервисах каждый сервис — отдельная база, отдельный процесс, отдельная сеть. Откатить чужую транзакцию нельзя. Сеть может упасть в самый неудобный момент.
Эта статья разбирает паттерны, которые решают именно эту проблему — один за другим, от простых к сложным.
Two-Phase Commit (2PC) — двухфазная фиксация
Проблема. Нужно, чтобы три сервиса либо все зафиксировали изменения, либо все откатились. Как это скоординировать?
Идея. Вводится координатор, который управляет процессом из двух фаз:
- Prepare (голосование): координатор спрашивает каждого участника «готов зафиксировать?». Участник выполняет всю работу — блокирует ресурсы, пишет в журнал — но не фиксирует. Отвечает «да» или «нет».
- Commit или Rollback (решение): если все ответили «да» — координатор говорит «фиксируй». Если хоть один ответил «нет» — все откатываются.
Почему 2PC редко используют в микросервисах:
- Между фазами ресурсы заблокированы. Если координатор упал — участники висят с заблокированными строками.
- Координатор — единая точка отказа. Его падение между фазами оставляет систему в неопределённом состоянии.
- Два сетевых раунда — задержка растёт с каждым участником.
Когда подходит: несколько баз данных одного вендора в рамках одной инфраструктуры (XA-транзакции). Для микросервисов через сеть — почти никогда.
Three-Phase Commit (3PC) — почему не спасает
3PC добавляет промежуточную фазу PRE_COMMIT, чтобы участники могли самостоятельно принять решение, если координатор пропадёт. Три сетевых раунда вместо двух — ещё больше задержка. И главное: 3PC не решает проблему сетевого раздела (network partition). Если участник получил PRE_COMMIT, потерял связь, зафиксировал — а другой участник в это время откатился — данные рассогласованы.
На практике проблемы 2PC решают не через 3PC, а через SAGA.
SAGA — цепочка компенсируемых шагов
Проблема. Нужно скоординировать действия нескольких сервисов, но без блокировок и единого координатора на уровне базы.
Идея. Вместо одной большой транзакции — цепочка локальных транзакций. Каждый шаг фиксируется самостоятельно. Если шаг N упал — выполняются компенсирующие транзакции для уже выполненных шагов в обратном порядке.
Шаг 1: Создать заказ → Компенсация: Отменить заказ
Шаг 2: Списать деньги → Компенсация: Вернуть деньги
Шаг 3: Зарезервировать на складе → Компенсация: Снять резерв
Ключевое отличие от 2PC: между шагами система находится во временно несогласованном состоянии. Это называется eventual consistency — в конечном счёте всё придёт к согласованности, но не мгновенно.
Оркестрация
Центральный оркестратор знает последовательность шагов и какую компенсацию вызвать при ошибке.
@dataclass(frozen=True)
class SagaStep:
name: str
action: Callable[[SagaContext], StepResult]
compensation: Callable[[SagaContext], None]
class CreateOrderSaga:
def execute(self, request: CreateOrderRequest) -> OrderResult:
ctx = SagaContext(request)
completed_steps: list[SagaStep] = []
steps = [
SagaStep("create-order",
lambda c: self.order_service.create(c.request),
lambda c: self.order_service.cancel(c.order_id)),
SagaStep("charge-payment",
lambda c: self.payment_service.charge(c.user_id, c.total),
lambda c: self.payment_service.refund(c.payment_id)),
SagaStep("reserve-inventory",
lambda c: self.inventory_service.reserve(c.order_id, c.items),
lambda c: self.inventory_service.release_reservation(c.order_id)),
]
for step in steps:
try:
result = step.action(ctx)
ctx.apply(result)
completed_steps.append(step)
except Exception as e:
self._compensate(ctx, completed_steps)
raise SagaError(f"Step failed: {step.name}") from e
return ctx.to_result()
Плюсы: логика в одном месте, легко отслеживать состояние, просто отлаживать.
Минусы: оркестратор знает про все сервисы — может разрастись в класс, который делает всё на свете.
Хореография
Нет центрального координатора. Каждый сервис слушает события и реагирует, публикуя свои:
# PaymentService реагирует на событие
async def on_order_created(event: OrderCreatedEvent) -> None:
try:
result = await payment_service.charge(event.user_id, event.total)
await publisher.publish(PaymentChargedEvent(event.order_id, result.payment_id))
except Exception as e:
await publisher.publish(PaymentFailedEvent(event.order_id, str(e)))
async def consume_order_events() -> None:
consumer = AIOKafkaConsumer("order-events", bootstrap_servers=KAFKA_URL)
await consumer.start()
try:
async for msg in consumer:
await on_order_created(OrderCreatedEvent.from_json(msg.value))
finally:
await consumer.stop()
Плюсы: сервисы слабо связаны, развиваются независимо.
Минусы: логику саги трудно отследить — она размазана по сервисам. Сложно понять, в каком состоянии находится процесс.
Оркестрация или хореография — что выбрать
Оркестрация лучше, когда шагов больше трёх-четырёх, есть ветвления («если оплата частичная — другой путь»), важна единая точка мониторинга.
Хореография лучше, когда шагов два-три, поток линейный без ветвлений, сервисы разрабатывают разные команды и им важна автономность.
Важно о компенсациях
Компенсация — это не откат базы данных. refund() — это новая бизнес-операция, которая приводит систему к эквиваленту отмены. Деньги возвращаются отдельной транзакцией, а не отменой списания.
Компенсация обязана быть идемпотентной — если вызвать её дважды, результат должен быть одинаковым:
def refund(self, payment_id: str) -> None:
payment = self.payment_repository.get(payment_id)
if payment.status == PaymentStatus.REFUNDED:
return # уже возвращено — ничего не делаем
self.payment_gateway.refund(payment.gateway_id)
payment.mark_refunded()
self.payment_repository.save(payment)
Transactional Outbox — атомарная запись данных и событий
Проблема. Типичная ошибка: сохранили заказ в базу, потом отправили событие в Kafka. Между этими двумя операциями процесс упал — заказ есть, событие потеряно. Обратная ситуация тоже плоха: событие ушло, а транзакция откатилась.
# Опасный код — не атомарно!
order_repository.save(order) # транзакция зафиксирована
await kafka.send(order_created) # процесс упал — событие потеряно
Решение. Событие сохраняется в ту же базу данных, в той же транзакции, что и бизнес-данные — в отдельную таблицу outbox_events. Отдельный фоновый процесс (relay) периодически читает эту таблицу и публикует события в брокер.
CREATE TABLE outbox_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
aggregate_type VARCHAR(255) NOT NULL,
aggregate_id VARCHAR(255) NOT NULL,
event_type VARCHAR(255) NOT NULL,
payload JSONB NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT now(),
published_at TIMESTAMP,
retry_count INT DEFAULT 0
);
# Бизнес-код: заказ и событие в одной транзакции
def create_order(self, cmd: CreateOrderCommand) -> Order:
with self.session.begin():
order = new_order(cmd)
self.session.add(order)
self.session.add(OutboxEvent(
aggregate_type="Order",
aggregate_id=str(order.id),
event_type="OrderCreated",
payload=order_created_payload(order),
))
return order
# Relay: периодически отправляет необработанные события
async def publish_pending_loop(self) -> None:
while True:
with self.session_factory.begin() as session:
events = find_unpublished(session, limit=100)
for event in events:
try:
await self.kafka.send_and_wait(
topic(event), event.payload, key=event.aggregate_id.encode())
event.mark_published()
except Exception:
event.retry_count += 1
await asyncio.sleep(0.5)
At-least-once — почему получатель должен быть идемпотентным
Outbox гарантирует at-least-once delivery: событие будет отправлено хотя бы один раз. Relay может упасть после отправки, но до того, как пометил строку published_at — тогда при следующем запуске событие уйдёт повторно. Поэтому потребители событий обязаны быть идемпотентными (см. раздел Idempotent Consumer ниже).
Polling vs CDC
Альтернатива polling-relay — Change Data Capture (CDC): внешний сервис читает журнал базы (WAL) и публикует изменения в Kafka напрямую. Но polling-relay в коде сервиса проще и надёжнее:
- Контракт события — в коде, а не в схеме таблицы. Изменение структуры
outbox_eventsне ломает потребителей. - CDC — отдельный кластер с собственным жизненным циклом и мониторингом. Сломался — отдельная команда, отдельная процедура.
- Висящий CDC-коннектор накапливает WAL до отказа диска. Polling-relay таких рисков не создаёт.
Опрос каждые 200–500 мс — задержка, неотличимая для бизнеса, и снимает целый класс операционных проблем.
Event Sourcing — история как источник истины
Проблема. Таблица orders хранит текущий статус: status = 'CONFIRMED'. Но мы не знаем, когда заказ был создан, когда оплачен, был ли он до этого в другом статусе. История потеряна.
Идея. Хранить не текущее состояние, а последовательность событий, которые к нему привели. Текущее состояние вычисляется воспроизведением всех событий.
OrderCreated → PaymentReceived → ItemReserved → OrderConfirmed
Агрегат восстанавливается из событий:
class Order:
def __init__(self) -> None:
self.id: str | None = None
self.status: OrderStatus | None = None
self.uncommitted_events: list[DomainEvent] = []
self.version = 0
@classmethod
def from_events(cls, events: list[DomainEvent]) -> "Order":
order = cls()
for event in events:
order._apply(event)
order.version += 1
return order
def _apply(self, event: DomainEvent) -> None:
match event:
case OrderCreated(order_id=order_id):
self.id = order_id
self.status = OrderStatus.CREATED
case PaymentReceived():
self.status = OrderStatus.PAID
case OrderConfirmed():
self.status = OrderStatus.CONFIRMED
case _:
raise ValueError(f"Unknown event: {type(event)}")
def confirm(self) -> None:
if self.status != OrderStatus.PAID:
raise RuntimeError(f"Can only confirm PAID orders, current={self.status}")
self._raise(OrderConfirmed(self.id, datetime.now(UTC)))
def _raise(self, event: DomainEvent) -> None:
self._apply(event)
self.uncommitted_events.append(event)
Проекции для чтения
Event Sourcing разделяет запись и чтение. Данные пишутся в Event Store (только добавление). Для чтения строятся проекции — таблицы, оптимизированные под конкретные запросы. Это естественно сочетается с CQRS.
class OrderSummaryProjection:
def on_order_created(self, event: OrderCreated) -> None:
self.repository.save(OrderSummary(event.order_id, "CREATED", event.created_at))
def on_order_confirmed(self, event: OrderConfirmed) -> None:
self.repository.update_status(event.order_id, "CONFIRMED")
Когда подходит Event Sourcing
Подходит, когда нужна полная история изменений (финансы, аудит, расчёты с продавцами), возможность посмотреть состояние на любой момент времени, или сложная доменная логика с множеством переходов статусов.
Избыточен для простых справочников, CRUD без сложной логики, проектов без требований к истории.
Подводные камни
- Задержка проекций. Проекция обновляется асинхронно — пользователь может не увидеть свои изменения сразу.
- Версионирование событий. Событие
OrderCreated_v1не содержит поляcurrency. В_v2оно появилось. Нужна стратегия миграции старых событий при чтении. - Размер Event Store. Агрегат с тысячами событий — воспроизведение занимает время. Решение — снимки (снэпшоты): периодически сохранять текущее состояние, воспроизводить только от последнего снимка.
Idempotent Consumer — защита от дублей
Проблема. В распределённой системе сообщения приходят повторно: ретраи брокера, перебалансировка группы потребителей, гарантии at-least-once. Обработчик должен давать одинаковый результат при повторной обработке.
Решение. Хранить таблицу обработанных событий. Перед обработкой проверить — не обрабатывали ли уже это событие.
def process(self, event_id: str, handler: Callable[[], None]) -> None:
with self.session.begin():
if self.session.get(ProcessedEvent, event_id) is not None:
return # уже обработали — пропускаем
handler()
self.session.add(ProcessedEvent(event_id=event_id, processed_at=datetime.now(UTC)))
# обработчик сообщений aiokafka из топика payment-events
async def on_payment_charged(event: PaymentChargedEvent) -> None:
def handle() -> None:
order = order_repository.get(event.order_id)
order.mark_paid(event.payment_id)
order_repository.save(order)
idempotent_processor.process(event.event_id, handle)
CREATE TABLE processed_events (
event_id VARCHAR(255) PRIMARY KEY,
processed_at TIMESTAMP NOT NULL DEFAULT now()
);
-- Очистка старых записей
DELETE FROM processed_events
WHERE processed_at < now() - INTERVAL '7 days';
Тот же принцип работает для REST API — через заголовок Idempotency-Key. Клиент генерирует уникальный ключ, сервер проверяет, не обрабатывался ли уже запрос с таким ключом:
@app.post("/payments")
async def charge(
request: ChargeRequest,
idempotency_key: Annotated[str, Header(alias="Idempotency-Key")],
) -> PaymentResult:
return idempotent_processor.process_or_return(
idempotency_key, lambda: payment_service.charge(request))
Distributed Lock — защита от конкурентного доступа
Проблема. Два экземпляра одного сервиса одновременно обрабатывают один заказ. Оба списывают деньги — пользователь платит дважды.
Решение. Распределённая блокировка через Redis: только один экземпляр может держать блокировку на один ресурс одновременно.
def execute_with_lock(self, lock_key: str, timeout: timedelta, action: Callable[[], T]) -> T:
lock = self.redis.lock(
lock_key,
timeout=timeout.total_seconds() * 2,
blocking_timeout=timeout.total_seconds(),
)
if not lock.acquire():
raise LockAcquisitionError(f"Cannot acquire lock: {lock_key}")
try:
return action()
finally:
if lock.owned():
lock.release()
def process_order(self, order_id: int) -> None:
def confirm() -> None:
order = self.order_repository.get(order_id)
order.confirm()
self.order_repository.save(order)
self.lock_service.execute_with_lock(
f"order-processing:{order_id}", timedelta(seconds=30), confirm)
Если Redis недоступен — можно использовать SELECT FOR UPDATE в базе данных:
def find_by_id_for_update(self, session: Session, order_id: int) -> Order | None:
row = session.execute(
select(OrderRow)
.where(OrderRow.id == order_id)
.with_for_update(skip_locked=True)
).scalar_one_or_none()
return to_domain_order(row) if row is not None else None
Подводные камни блокировок:
- Взаимная блокировка. Два процесса блокируют ресурсы в разном порядке и ждут друг друга. Решение — всегда блокировать в одном порядке (например, по ID).
- Зависшая блокировка. Процесс взял блокировку, упал, не отпустил. Решение — TTL (автоматическое снятие по таймауту).
- Раздвоение в Redis. Redis перешёл на нового мастера, а старый ещё жив — два процесса держат одну блокировку. Решение — Redlock (блокировка сразу на нескольких независимых нодах Redis).
Как паттерны работают вместе
Реальная система не использует один паттерн в изоляции. Они складываются в цепочку:
OrderService создаёт заказ и сохраняет событие через Outbox — атомарно в одной транзакции. Polling Relay периодически читает таблицу и публикует в Kafka. PaymentService обрабатывает через Idempotent Consumer — дубли безопасны. При конкурентном доступе — Distributed Lock. Весь процесс координирует SAGA с компенсациями, если что-то пошло не так.
Коротко
- 2PC — строгая согласованность через двухфазный протокол; подходит только в рамках одной инфраструктуры, для микросервисов через сеть не применяют.
- SAGA — цепочка локальных транзакций с компенсирующими шагами; оркестрация (центральный координатор) или хореография (события между сервисами).
- Компенсация — это новая бизнес-операция, а не откат базы; обязана быть идемпотентной.
- Transactional Outbox — событие и бизнес-данные в одной транзакции; relay доставляет в брокер с гарантией at-least-once.
- Event Sourcing — хранение истории событий вместо текущего состояния; восстановление через воспроизведение; сочетается с CQRS.
- Idempotent Consumer — таблица обработанных событий защищает от дублей при повторной доставке.
- Distributed Lock — Redis-блокировка (redis-py/Redlock) или
SELECT FOR UPDATEзащищают от конкурентного доступа.
Что почитать дальше
- CQRS — паттерн разделения команд и запросов; естественно сочетается с Event Sourcing.
- Гексагональная архитектура — как изолировать SAGA-оркестратор от инфраструктуры.