Когда приложение разбивают на несколько сервисов, исчезает главное удобство монолита — единая транзакция базы данных. Раньше одна транзакция базы данных гарантировала: либо заказ создан, деньги списаны и склад зарезервирован — либо ничего из этого. В микросервисах каждый сервис — отдельная база, отдельный процесс, отдельная сеть. Откатить чужую транзакцию нельзя. Сеть может упасть в самый неудобный момент.
Эта статья разбирает паттерны, которые решают именно эту проблему — один за другим, от простых к сложным.
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 — в конечном счёте всё придёт к согласованности, но не мгновенно.
Оркестрация
Центральный оркестратор знает последовательность шагов и какую компенсацию вызвать при ошибке.
type SagaStep struct {
Name string
Action func(ctx context.Context, s *SagaState) error
Compensate func(ctx context.Context, s *SagaState) error
}
func (s *CreateOrderSaga) Execute(ctx context.Context, req CreateOrderRequest) (OrderResult, error) {
state := NewSagaState(req)
steps := []SagaStep{
{
Name: "create-order",
Action: func(ctx context.Context, st *SagaState) error { return s.orders.Create(ctx, st) },
Compensate: func(ctx context.Context, st *SagaState) error { return s.orders.Cancel(ctx, st.OrderID) },
},
{
Name: "charge-payment",
Action: func(ctx context.Context, st *SagaState) error { return s.payments.Charge(ctx, st) },
Compensate: func(ctx context.Context, st *SagaState) error { return s.payments.Refund(ctx, st.PaymentID) },
},
{
Name: "reserve-inventory",
Action: func(ctx context.Context, st *SagaState) error { return s.inventory.Reserve(ctx, st) },
Compensate: func(ctx context.Context, st *SagaState) error { return s.inventory.Release(ctx, st.OrderID) },
},
}
var completed []SagaStep
for _, step := range steps {
if err := step.Action(ctx, state); err != nil {
s.compensate(ctx, state, completed)
return OrderResult{}, fmt.Errorf("saga step %s: %w", step.Name, err)
}
completed = append(completed, step)
}
return state.Result(), nil
}
Плюсы: логика в одном месте, легко отслеживать состояние, просто отлаживать.
Минусы: оркестратор знает про все сервисы — может разрастись в структуру, которая делает всё на свете.
Хореография
Нет центрального координатора. Каждый сервис слушает события и реагирует, публикуя свои:
// PaymentService реагирует на событие
func (s *PaymentService) ConsumeOrderEvents(ctx context.Context) error {
reader := kafka.NewReader(kafka.ReaderConfig{
Brokers: s.brokers,
GroupID: "payment-service",
Topic: "order-events",
})
defer reader.Close()
for {
msg, err := reader.ReadMessage(ctx)
if err != nil {
return err
}
var event OrderCreatedEvent
if err := json.Unmarshal(msg.Value, &event); err != nil {
continue
}
result, err := s.charge(ctx, event.UserID, event.Total)
if err != nil {
s.publish(ctx, PaymentFailedEvent{OrderID: event.OrderID, Reason: err.Error()})
continue
}
s.publish(ctx, PaymentChargedEvent{OrderID: event.OrderID, PaymentID: result.PaymentID})
}
}
Плюсы: сервисы слабо связаны, развиваются независимо.
Минусы: логику саги трудно отследить — она размазана по сервисам. Сложно понять, в каком состоянии находится процесс.
Оркестрация или хореография — что выбрать
Оркестрация лучше, когда шагов больше трёх-четырёх, есть ветвления («если оплата частичная — другой путь»), важна единая точка мониторинга.
Хореография лучше, когда шагов два-три, поток линейный без ветвлений, сервисы разрабатывают разные команды и им важна автономность.
Важно о компенсациях
Компенсация — это не откат базы данных. Refund() — это новая бизнес-операция, которая приводит систему к эквиваленту отмены. Деньги возвращаются отдельной транзакцией, а не отменой списания.
Компенсация обязана быть идемпотентной — если вызвать её дважды, результат должен быть одинаковым:
func (s *PaymentService) Refund(ctx context.Context, paymentID string) error {
payment, err := s.repo.FindByID(ctx, paymentID)
if err != nil {
return err
}
if payment.Status == StatusRefunded {
return nil // уже возвращено — ничего не делаем
}
if err := s.gateway.Refund(ctx, payment.GatewayID); err != nil {
return err
}
payment.MarkRefunded()
return s.repo.Save(ctx, payment)
}
Transactional Outbox — атомарная запись данных и событий
Проблема. Типичная ошибка: сохранили заказ в базу, потом отправили событие в Kafka. Между этими двумя операциями процесс упал — заказ есть, событие потеряно. Обратная ситуация тоже плоха: событие ушло, а транзакция откатилась.
// Опасный код — не атомарно!
repo.Save(ctx, order) // транзакция зафиксирована
writer.WriteMessages(ctx, orderCreated) // процесс упал — событие потеряно
Решение. Событие сохраняется в ту же базу данных, в той же транзакции, что и бизнес-данные — в отдельную таблицу 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
);
// Бизнес-код: заказ и событие в одной транзакции
func (s *OrderService) CreateOrder(ctx context.Context, cmd CreateOrderCommand) (Order, error) {
var order Order
err := pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
var err error
order, err = s.orders.SaveTx(ctx, tx, newOrder(cmd))
if err != nil {
return err
}
return s.outbox.SaveTx(ctx, tx,
"Order", order.ID, "OrderCreated", NewOrderCreatedPayload(order))
})
return order, err
}
// Relay: периодически отправляет необработанные события
func (r *OutboxRelay) Run(ctx context.Context) {
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
r.publishPending(ctx)
}
}
}
func (r *OutboxRelay) publishPending(ctx context.Context) {
events, err := r.repo.FindUnpublished(ctx, 100)
if err != nil {
return
}
for _, event := range events {
err := r.writer.WriteMessages(ctx, kafka.Message{
Topic: topicFor(event),
Key: []byte(event.AggregateID),
Value: event.Payload,
})
if err != nil {
event.RetryCount++
} else {
event.MarkPublished()
}
r.repo.Save(ctx, event)
}
}
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
Агрегат восстанавливается из событий:
type Order struct {
ID string
Status OrderStatus
uncommittedEvents []DomainEvent
version int64
}
func OrderFromEvents(events []DomainEvent) (*Order, error) {
order := &Order{}
for _, event := range events {
if err := order.apply(event); err != nil {
return nil, err
}
order.version++
}
return order, nil
}
func (o *Order) apply(event DomainEvent) error {
switch e := event.(type) {
case OrderCreated:
o.ID = e.OrderID
o.Status = StatusCreated
case PaymentReceived:
o.Status = StatusPaid
case OrderConfirmed:
o.Status = StatusConfirmed
default:
return fmt.Errorf("unknown event: %T", e)
}
return nil
}
func (o *Order) Confirm() error {
if o.Status != StatusPaid {
return fmt.Errorf("can only confirm PAID orders, current=%s", o.Status)
}
return o.raise(OrderConfirmed{OrderID: o.ID, At: time.Now()})
}
func (o *Order) raise(event DomainEvent) error {
if err := o.apply(event); err != nil {
return err
}
o.uncommittedEvents = append(o.uncommittedEvents, event)
return nil
}
Проекции для чтения
Event Sourcing разделяет запись и чтение. Данные пишутся в Event Store (только добавление). Для чтения строятся проекции — таблицы, оптимизированные под конкретные запросы. Это естественно сочетается с CQRS.
func (p *OrderSummaryProjection) OnOrderCreated(ctx context.Context, event OrderCreated) error {
return p.repo.Save(ctx, OrderSummary{
OrderID: event.OrderID,
Status: "CREATED",
CreatedAt: event.CreatedAt,
})
}
func (p *OrderSummaryProjection) OnOrderConfirmed(ctx context.Context, event OrderConfirmed) error {
return p.repo.UpdateStatus(ctx, event.OrderID, "CONFIRMED")
}
Когда подходит Event Sourcing
Подходит, когда нужна полная история изменений (финансы, аудит, расчёты с продавцами), возможность посмотреть состояние на любой момент времени, или сложная доменная логика с множеством переходов статусов.
Избыточен для простых справочников, CRUD без сложной логики, проектов без требований к истории.
Подводные камни
- Задержка проекций. Проекция обновляется асинхронно — пользователь может не увидеть свои изменения сразу.
- Версионирование событий. Событие
OrderCreated_v1не содержит поляcurrency. В_v2оно появилось. Нужна стратегия миграции старых событий при чтении. - Размер Event Store. Агрегат с тысячами событий — воспроизведение занимает время. Решение — снимки (снэпшоты): периодически сохранять текущее состояние, воспроизводить только от последнего снимка.
Idempotent Consumer — защита от дублей
Проблема. В распределённой системе сообщения приходят повторно: ретраи брокера, перебалансировка группы потребителей, гарантии at-least-once. Обработчик должен давать одинаковый результат при повторной обработке.
Решение. Хранить таблицу обработанных событий. Перед обработкой проверить — не обрабатывали ли уже это событие.
func (p *IdempotentProcessor) Process(ctx context.Context, eventID string,
handler func(ctx context.Context, tx pgx.Tx) error) error {
return pgx.BeginFunc(ctx, p.pool, func(tx pgx.Tx) error {
var exists bool
err := tx.QueryRow(ctx,
`SELECT EXISTS(SELECT 1 FROM processed_events WHERE event_id = $1)`,
eventID).Scan(&exists)
if err != nil {
return err
}
if exists {
return nil // уже обработали — пропускаем
}
if err := handler(ctx, tx); err != nil {
return err
}
_, err = tx.Exec(ctx,
`INSERT INTO processed_events (event_id, processed_at) VALUES ($1, now())`,
eventID)
return err
})
}
func (c *PaymentEventsConsumer) Run(ctx context.Context) error {
for {
msg, err := c.reader.ReadMessage(ctx)
if err != nil {
return err
}
var event PaymentChargedEvent
if err := json.Unmarshal(msg.Value, &event); err != nil {
continue
}
c.processor.Process(ctx, event.EventID, func(ctx context.Context, tx pgx.Tx) error {
order, err := c.orders.FindByIDTx(ctx, tx, event.OrderID)
if err != nil {
return err
}
order.MarkPaid(event.PaymentID)
return c.orders.SaveTx(ctx, tx, order)
})
}
}
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. Клиент генерирует уникальный ключ, сервер проверяет, не обрабатывался ли уже запрос с таким ключом:
func (h *PaymentHandler) Charge(w http.ResponseWriter, r *http.Request) {
idempotencyKey := r.Header.Get("Idempotency-Key")
var req ChargeRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
result, err := h.processor.ProcessOrReturn(r.Context(), idempotencyKey,
func(ctx context.Context) (PaymentResult, error) {
return h.payments.Charge(ctx, req)
})
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
json.NewEncoder(w).Encode(result)
}
Distributed Lock — защита от конкурентного доступа
Проблема. Два экземпляра одного сервиса одновременно обрабатывают один заказ. Оба списывают деньги — пользователь платит дважды.
Решение. Распределённая блокировка через Redis: только один экземпляр может держать блокировку на один ресурс одновременно.
func (l *LockService) ExecuteWithLock(ctx context.Context, lockKey string,
ttl time.Duration, action func(ctx context.Context) error) error {
mutex := l.redsync.NewMutex(lockKey, redsync.WithExpiry(ttl))
if err := mutex.LockContext(ctx); err != nil {
return fmt.Errorf("cannot acquire lock %s: %w", lockKey, err)
}
defer mutex.UnlockContext(ctx)
return action(ctx)
}
func (s *OrderProcessor) ProcessOrder(ctx context.Context, orderID int64) error {
return s.locks.ExecuteWithLock(ctx,
fmt.Sprintf("order-processing:%d", orderID),
30*time.Second,
func(ctx context.Context) error {
order, err := s.orders.FindByID(ctx, orderID)
if err != nil {
return err
}
if err := order.Confirm(); err != nil {
return err
}
return s.orders.Save(ctx, order)
})
}
Если Redis недоступен — можно использовать SELECT FOR UPDATE в базе данных:
func (r *OrderRepository) FindByIDForUpdate(ctx context.Context, tx pgx.Tx, id int64) (*Order, error) {
row := tx.QueryRow(ctx, `
SELECT id, status, total
FROM orders
WHERE id = $1
FOR UPDATE SKIP LOCKED`, id)
var order Order
if err := row.Scan(&order.ID, &order.Status, &order.Total); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
return nil, err
}
return &order, nil
}
Подводные камни блокировок:
- Взаимная блокировка. Два процесса блокируют ресурсы в разном порядке и ждут друг друга. Решение — всегда блокировать в одном порядке (например, по 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-блокировка (redsync/Redlock) или
SELECT FOR UPDATEзащищают от конкурентного доступа.
Что почитать дальше
- CQRS — паттерн разделения команд и запросов; естественно сочетается с Event Sourcing.
- Гексагональная архитектура — как изолировать SAGA-оркестратор от инфраструктуры.