Перейти к содержимому
Подготовка к собеседованию

Очереди и Kafka

Как сервисы обмениваются сообщениями асинхронно: чем очередь отличается от лога событий, как устроена Kafka, какие гарантии доставки реальны и как не потерять событие между базой и брокером. Без этих понятий не обсудить ни одну интеграцию между сервисами, поэтому их спрашивают почти на каждом собеседовании по бэкенду.

Перед этой темой: Стили интеграции, Транзакции и блокировки.

Junior+

Чем очередь сообщений (RabbitMQ, Azure Service Bus) отличается от лога событий (Kafka)?

Заголовок раздела «Чем очередь сообщений (RabbitMQ, Azure Service Bus) отличается от лога событий (Kafka)?»
Ответ

Что это

  • Очередь сообщений — почтовый ящик: получатель забирает сообщение, подтверждает обработку, и брокер его удаляет.
  • Лог событий — журнал, в который события только дописываются. Читатели идут по нему каждый в своём темпе, а события хранятся заданный срок.

Как это проявляется

Признак Очередь Лог
После обработки сообщение удаляется остаётся до конца срока хранения
Старые сообщения перечитать нельзя можно с любой позиции
Новый получатель видит только новые может прочитать историю

Зачем вообще брокер

Он развязывает сервисы: отправитель не ждёт получателя, а сообщения не теряются, пока получатель перезапускается.

Частая ошибка

Считать, что Kafka удаляет сообщение после чтения. Оно остаётся в топике, а «прочитано» — это только сохранённая позиция конкретного читателя.

Как устроено

  • Point-to-point. Несколько экземпляров сервиса читают одну очередь и конкурируют: каждое сообщение достаётся одному. Так масштабируют обработку.
  • Pub/sub в очереди. Exchange в RabbitMQ копирует сообщение во все привязанные очереди, топик в Service Bus — во все подписки. У каждого подписчика своя копия.
  • Pub/sub в логе. Запись одна, а читают её независимые группы потребителей, у каждой своя позиция. Внутри группы экземпляры делят между собой партиции — части топика.

Когда что выбирать

Задача Выбор
Команда одному исполнителю, фоновое задание очередь
Повтор с задержкой, приоритет, TTL сообщения очередь
Факт «произошло» для многих систем, перечитывание лог
Большой поток событий с порядком по ключу лог

Ловушки

  • Kafka как очередь задач. Одно медленное сообщение задерживает всю партицию, а обработчиков в группе не больше, чем партиций.
  • Очередь как история. Новый потребитель не прочитает прошлые события: их уже удалили.

Правило

Команды и задания — в очередь, события-факты для многих потребителей — в лог.

Под капотом

  • Очередь хранит состояние каждого сообщения: кому выдано, сколько раз доставлялось, когда истечёт блокировка. Подтверждённое удаляется, поэтому очередь короткая, пока потребители успевают.
  • Лог — файлы-сегменты, которые брокер пишет и читает последовательно. Прогресс потребителя — одно число на партицию, и брокер не тратит ресурсы на учёт отдельных сообщений.
  • Данные из лога удаляет не чтение, а политика хранения: по времени, по размеру или уплотнение по ключу.

Компромиссы

Свойство Очередь Лог
Учёт доставки на сообщение: повтор, задержка, приоритет на партицию: одна позиция
Масштабирование чтения добавить обработчиков не больше потребителей, чем партиций
Порядок теряется при конкурирующих получателях гарантирован внутри партиции
История нет перечитывание, новые потребители

Границы размываются

  • RabbitMQ Streams — лог внутри RabbitMQ: сообщения не удаляются при чтении, их можно перечитать.
  • В Kafka 4.2 share groups (KIP-932, «очереди для Kafka») стали production-ready: несколько потребителей читают одну партицию и подтверждают записи по одной.

Как проверить в проде

  • Очередь: длина и возраст самого старого сообщения показывают, что обработчики не успевают.
  • Лог: отставание группы (consumer lag) — сколько записей между концом партиции и позицией группы.

Итог: выбор между очередью и логом — выбор, кто отвечает за доставку: брокер за каждое сообщение или потребитель за свою позицию в потоке.

Пример
Confluent.Kafka
using Confluent.Kafka;


// Две группы читают один топик независимо: каждая получит все события
using var billing = CreateConsumer("billing");
using var analytics = CreateConsumer("analytics");
billing.Subscribe("orders");
analytics.Subscribe("orders");


// Лог можно перечитать: встаём в начало партиции 0 без участия отправителя
using var replay = CreateConsumer("replay");
replay.Assign(new TopicPartitionOffset("orders", 0, Offset.Beginning));
var first = replay.Consume(TimeSpan.FromSeconds(5));
Console.WriteLine(first?.Message.Value); // самое старое событие, которое ещё хранится


static IConsumer<string, string> CreateConsumer(string groupId) =>
    new ConsumerBuilder<string, string>(new ConsumerConfig
    {
        BootstrapServers = "localhost:9092",
        GroupId = groupId,
        AutoOffsetReset = AutoOffsetReset.Earliest // новая группа начинает с самого старого события
    }).Build();

В очереди так не получится: обработанное сообщение уже удалено, а новая подписка видит только то, что пришло после её создания.

Источники
Middle+

Как устроена Kafka?

Заголовок раздела «Как устроена Kafka?»
Ответ

Что это

  • Топик — именованный поток событий, например orders.
  • Партиция — часть топика, отдельный упорядоченный лог. Каждая запись в партиции получает номер — offset.
  • Producer пишет записи, consumer читает их.
  • Группа потребителей — экземпляры одного сервиса с общим GroupId. Партиции топика делятся между ними.

Как пользоваться

  1. Producer отправляет запись с ключом, например с id заказа.
  2. Kafka выбирает партицию по хешу ключа: события одного заказа всегда попадают в одну партицию и идут по порядку.
  3. Consumer читает свои партиции и сохраняет offset — докуда обработал.

Частая ошибка

Ждать порядка во всём топике. Порядок есть только внутри партиции, а записи разных партиций читаются вперемешку.

Как устроено

  • Ключ. Запись с ключом попадает в партицию по хешу ключа, без ключа — распределяется для баланса.
  • Группа. Одну партицию в группе читает один потребитель; если потребителей больше, чем партиций, лишние простаивают.
  • Ребалансировка. Партиции перераспределяются, когда участник пришёл, ушёл, не ответил за session.timeout.ms или не вызывал Consume дольше max.poll.interval.ms.
  • Offset. Группа коммитит offset следующей непрочитанной записи во внутренний топик __consumer_offsets.
  • Реплики. У партиции есть лидер и последователи на других брокерах. Синхронные реплики (ISR) не отстают от лидера.
  • Хранение. cleanup.policy=delete удаляет старые сегменты по времени или размеру (по умолчанию через 7 дней), compact оставляет последнюю запись каждого ключа — уплотнение лога.

Когда что выбирать

Настройка acks Что даёт
0 максимум скорости, потери возможны
1 подтверждает лидер; потеря, если он упадёт до репликации
all и min.insync.replicas=2 запись переживёт падение брокера

Ловушки

Долгая обработка между вызовами Consume выкидывает потребителя из группы: партиции уходят другому, и незакоммиченные записи обрабатываются повторно.

Правило

Ключ — сущность, для которой важен порядок; партиций — с запасом под будущий параллелизм.

Под капотом

  • Партиция на диске — набор файлов-сегментов. Запись дописывается в конец, чтение идёт последовательно и опирается на page cache ОС, поэтому Kafka держит большой поток на обычных дисках.
  • Реплика, которая отстала дольше replica.lag.time.max.ms, выпадает из ISR. Если ISR меньше min.insync.replicas, запись с acks=all получает ошибку NotEnoughReplicas.
  • С Kafka 4.0 метаданные хранит кворум контроллеров KRaft, режима ZooKeeper больше нет.
  • Протокол групп KIP-848 (GA в Kafka 4.0, в librdkafka — с 2.12) переносит назначение партиций на брокер. Включается group.protocol=consumer, по умолчанию в librdkafka пока classic.

Цена

  • Число партиций нельзя уменьшить. Увеличение меняет соответствие ключ → партиция: новые события ключа пойдут в другую партицию раньше, чем дочитаны старые.
  • Ключ с большой долей трафика, например крупный магазин, перегружает одну партицию.
  • Confluent.Kafka (librdkafka) по умолчанию хеширует ключ CRC32, а Java-клиент — murmur2. Producer-ы на разных языках раскладывают один ключ по разным партициям; для совместимости ставят Partitioner.Murmur2Random.

Как проверить в проде

  • kafka-consumer-groups.sh --describe --group billing показывает offset группы и отставание по каждой партиции.
  • Частые ребалансировки в логах клиента — признак долгой обработки или нестабильных экземпляров.

Итог: Kafka быстра, потому что брокер только хранит и реплицирует лог; ключ, число партиций и acks — решения приложения, и ошибки в них дорого исправлять.

Producer: ключ и порядок
Confluent.Kafka
using Confluent.Kafka;


var config = new ProducerConfig
{
    BootstrapServers = "localhost:9092",
    Acks = Acks.All,                        // ждать все синхронные реплики
    Partitioner = Partitioner.Murmur2Random // те же партиции, что у Java-клиента
};


using var producer = new ProducerBuilder<string, string>(config).Build();


foreach (var status in new[] { "Created", "Paid", "Shipped" })
{
    // Один ключ — одна партиция: события заказа 42 будут прочитаны по порядку
    var result = await producer.ProduceAsync("orders",
        new Message<string, string> { Key = "order-42", Value = status });
    Console.WriteLine($"{status}: partition {result.Partition.Value}, offset {result.Offset.Value}");
}
// Created: partition 3, offset 120 — номера для примера, партиция у всех трёх одна
// Paid: partition 3, offset 121
// Shipped: partition 3, offset 122
Consumer: группа и ребалансировка
Confluent.Kafka
using Confluent.Kafka;


var config = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "billing",                       // экземпляры с одним GroupId делят партиции
    AutoOffsetReset = AutoOffsetReset.Earliest // откуда читать, если у группы ещё нет offset
};


using var consumer = new ConsumerBuilder<string, string>(config)
    .SetPartitionsAssignedHandler((_, parts) => Console.WriteLine($"Assigned: {string.Join(", ", parts)}"))
    .SetPartitionsRevokedHandler((_, parts) => Console.WriteLine($"Revoked: {string.Join(", ", parts)}"))
    .Build();


consumer.Subscribe("orders");
var result = consumer.Consume(TimeSpan.FromSeconds(10)); // Assigned: orders [[0]], orders [[1]]
if (result is not null)
    Console.WriteLine($"{result.Message.Key}: {result.Message.Value} at {result.TopicPartitionOffset}");


consumer.Close(); // выйти из группы сразу, а не ждать session.timeout.ms

Запустите второй экземпляр с тем же GroupId — первый напечатает Revoked и получит половину партиций.

Источники
Middle+

Какие бывают гарантии доставки и как обрабатывать сбои?

Заголовок раздела «Какие бывают гарантии доставки и как обрабатывать сбои?»
Ответ

Что это

Гарантия Потеря Дубли
at-most-once возможна нет
at-least-once нет возможны
exactly-once нет нет

Откуда берутся дубли

  1. Потребитель обработал сообщение, например списал деньги.
  2. Он упал раньше, чем подтвердил обработку брокеру.
  3. После перезапуска брокер выдаёт это сообщение снова.

Как пользоваться

  • Подтверждайте сообщение после обработки, а не до.
  • Делайте обработку идемпотентной: проверяйте, не обработано ли уже сообщение с этим id.
  • Сообщение, которое падает раз за разом, после нескольких попыток откладывайте в dead letter queue — отдельную очередь для разбора.

Частая ошибка

Подтверждать сообщение сразу при получении. Если обработка упадёт, сообщение потеряно.

Как устроено

  • At-most-once. Подтверждение или коммит offset до обработки. Падение посреди обработки — сообщение потеряно.
  • At-least-once. Подтверждение после обработки. Падение между обработкой и подтверждением — повтор.
  • Exactly-once в Kafka. Идемпотентный producer отбрасывает дубли от повторных отправок. Транзакции атомарно объединяют запись в выходные топики и коммит offset, а потребители с read_committed не видят отменённых записей. Внешняя база в транзакцию не входит.

Как обрабатывать сбои

  1. Временная ошибка (таймаут, дедлок) — несколько повторов с паузой прямо в обработчике.
  2. Не помогло — отложенная повторная доставка или retry-топик с растущей задержкой.
  3. Попытки исчерпаны или ошибка постоянная (битый JSON, нарушен контракт) — в DLQ с причиной ошибки.

Ловушки

  • В Confluent.Kafka по умолчанию включены EnableAutoCommit и EnableAutoOffsetStore: offset считается обработанным, как только Consume вернул сообщение. Для at-least-once ставят EnableAutoOffsetStore = false и вызывают StoreOffset после обработки.
  • Без DLQ poison message в Kafka останавливает партицию: потребитель повторяет его бесконечно.

Правило

At-least-once, идемпотентный обработчик, ограниченное число повторов и DLQ с алертом.

Под капотом

  • Идемпотентный producer получает producer id, а брокер помнит номера последовательности по каждой партиции и отбрасывает повтор. Порядок сохраняется при max.in.flight.requests.per.connection до 5.
  • В Java-клиенте идемпотентность включена по умолчанию с Kafka 3.0, в librdkafka и Confluent.Kafka — нет: EnableIdempotence = true ставят явно.
  • Транзакционный producer с TransactionalId пишет записи и offset группы (SendOffsetsToTransaction) атомарно. librdkafka по умолчанию читает с read_committed, Java-клиент — с read_uncommitted.

Компромиссы

Повтор Плюс Минус
На месте, в обработчике порядок сохранён партиция стоит, пока идут повторы
Retry-топик и DLQ поток не блокируется порядок по ключу нарушается
Парковка ключа порядок по ключу сохранён нужно хранить «застрявшие» ключи

Транзакции Kafka добавляют задержку и маркеры коммита в лог, а внешние эффекты всё равно требуют дедупликации. Поэтому чаще выбирают at-least-once и идемпотентного потребителя.

Как проверить в проде

  • Размер и возраст DLQ — метрика с алертом: DLQ, которую никто не разбирает, — та же потеря данных, только отложенная.
  • Счётчики повторов и отброшенных дублей показывают, где обработчик не справляется.

Итог: exactly-once существует только внутри Kafka; на границе с внешним миром надёжность собирают из at-least-once, идемпотентности, ограниченных повторов и DLQ.

Пример
Confluent.Kafka
using System.Text;
using Confluent.Kafka;


var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "billing",
    EnableAutoCommit = true,      // коммит по таймеру...
    EnableAutoOffsetStore = false // ...только тех offset, что сохранены явно
};
var producerConfig = new ProducerConfig { BootstrapServers = "localhost:9092", EnableIdempotence = true };


using var consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
using var dlq = new ProducerBuilder<string, string>(producerConfig).Build();
consumer.Subscribe("orders");


while (true)
{
    var result = consumer.Consume();
    for (var attempt = 1; ; attempt++)
    {
        try { Handle(result.Message); break; }
        catch (Exception) when (attempt < 3) { await Task.Delay(200 * attempt); } // временный сбой
        catch (Exception ex)
        {
            // Poison message: откладываем в DLQ с причиной, чтобы не стоять на месте
            await dlq.ProduceAsync("orders.dlq", new Message<string, string>
            {
                Key = result.Message.Key,
                Value = result.Message.Value,
                Headers = new Headers { { "error", Encoding.UTF8.GetBytes(ex.Message) } }
            });
            break;
        }
    }
    consumer.StoreOffset(result); // обработано или в DLQ — только теперь offset можно коммитить
}


static void Handle(Message<string, string> message) { /* идемпотентная обработка */ }

Упадёт процесс до StoreOffset — сообщение придёт снова: это at-least-once, поэтому Handle обязан быть идемпотентным.

Источники
Senior+

Как надёжно опубликовать событие после изменения в БД?

Заголовок раздела «Как надёжно опубликовать событие после изменения в БД?»
Ответ

Что это

Сервис сохраняет заказ и публикует событие OrderCreated. Это две записи в разные системы — базу и брокер, общей транзакции у них нет.

Что может пойти не так

  • Заказ сохранён, а процесс упал до отправки — событие потеряно.
  • Событие отправлено, а транзакция в базе откатилась — другие сервисы узнали о несуществующем заказе.

Как пользоваться outbox

  1. В той же транзакции, что и заказ, вставьте строку в таблицу OutboxMessages: тип события и JSON.
  2. Фоновый процесс читает неотправленные строки и публикует их в брокер.
  3. После подтверждения брокера строка помечается отправленной.

Частая ошибка

Отправлять событие внутри транзакции БД и ждать, что откат его отменит. Брокер в транзакцию не входит.

Как устроено

  1. Бизнес-изменение и запись в outbox — одна локальная транзакция. В EF Core это один SaveChanges.
  2. Relay — BackgroundService или отдельный процесс — выбирает неотправленные записи по порядку.
  3. Публикует каждую в брокер и ждёт подтверждения.
  4. Помечает запись отправленной. Упал между шагами 3 и 4 — запись уйдёт повторно, это at-least-once.

Inbox на стороне потребителя

В одной транзакции потребитель вставляет id сообщения в таблицу InboxMessages с первичным ключом по id и применяет изменения. Повтор нарушит ключ: сообщение уже обработано, его подтверждают и пропускают.

Когда что выбирать

Доставка из outbox Когда
Опрос таблицы по умолчанию: умеренный поток, задержка в секунды допустима
CDC (Debezium) большой поток, нужна малая задержка, Kafka Connect уже есть

Ловушки

  • Несколько экземпляров relay без блокировок публикуют одни и те же записи. Строки захватывают через UPDLOCK, READPAST (SQL Server) или FOR UPDATE SKIP LOCKED (PostgreSQL) либо оставляют один экземпляр relay.
  • Таблица outbox растёт бесконечно, если не удалять отправленные записи.

Правило

Outbox — у отправителя, inbox или идемпотентный обработчик — у получателя.

Под капотом

  • Outbox не убирает распределённую транзакцию, а заменяет её локальной плюс at-least-once доставкой. Между сервисами остаётся согласованность в конечном счёте: подписчики узнают об изменении с задержкой.
  • CDC читает журнал транзакций: коннектор Debezium для SQL Server опирается на встроенный CDC, для PostgreSQL — на логическое декодирование. Outbox Event Router превращает строки outbox в записи Kafka с ключом агрегата.

Компромиссы

Relay Плюс Минус
Опрос таблицы только база и свой код нагрузка на базу, задержка до интервала опроса
CDC порядок коммитов, малая задержка Kafka Connect и доступ к журналу

Параллельный relay или повтор одной записи меняет порядок событий. Если порядок по агрегату важен, публикуют последовательно с ключом агрегата, а потребитель сверяет версию агрегата и отбрасывает устаревшее.

Готовые реализации

  • MassTransit: bus outbox для публикации вне потребителя и consumer outbox с inbox. С v9 MassTransit — коммерческий продукт, v8 остаётся open source.
  • NServiceBus Outbox: дедупликация входящих и отложенная отправка исходящих в одной транзакции; продукт коммерческий.
  • Wolverine: durable inbox и outbox поверх EF Core, Marten, SQL Server или PostgreSQL.

Как проверить в проде

Возраст самой старой неотправленной записи outbox — главная метрика: рост значит, что relay стоит или не успевает.

Итог: outbox и inbox заменяют распределённую транзакцию локальными; остаётся выбрать доставку — опрос или CDC — и решить, как сохранить порядок.

Outbox на EF Core: запись и один проход relay
// NuGet: Microsoft.EntityFrameworkCore.SqlServer, Confluent.Kafka
using System.Text.Json;
using Confluent.Kafka;
using Microsoft.EntityFrameworkCore;


await using var db = new ShopDb();


// 1. Заказ и событие — одна транзакция: сохранятся оба или ни одного
var order = new Order { Id = Guid.NewGuid(), Total = 150m };
db.Orders.Add(order);
db.Outbox.Add(new OutboxMessage
{
    MessageId = Guid.NewGuid(),
    AggregateId = order.Id.ToString(),
    Type = "OrderCreated",
    Payload = JsonSerializer.Serialize(new { order.Id, order.Total })
});
await db.SaveChangesAsync();


// 2. Relay: публикует неотправленное по порядку и помечает отправленным
using var producer = new ProducerBuilder<string, string>(
    new ProducerConfig { BootstrapServers = "localhost:9092", EnableIdempotence = true }).Build();
var pending = await db.Outbox.Where(m => m.SentAt == null).OrderBy(m => m.Id).Take(100).ToListAsync();
foreach (var message in pending)
{
    await producer.ProduceAsync("orders", new Message<string, string>
    {
        Key = message.AggregateId, // ключ агрегата — порядок событий одного заказа
        Value = message.Payload,
        Headers = new Headers { { "message-id", message.MessageId.ToByteArray() } }
    });
    message.SentAt = DateTime.UtcNow;
    await db.SaveChangesAsync(); // упадём до этой строки — событие уйдёт повторно
}


public class Order { public Guid Id { get; set; } public decimal Total { get; set; } }


public class OutboxMessage
{
    public long Id { get; set; }               // identity: задаёт порядок
    public Guid MessageId { get; set; }        // для дедупликации у потребителя
    public required string AggregateId { get; set; }
    public required string Type { get; set; }
    public required string Payload { get; set; }
    public DateTime? SentAt { get; set; }      // null — ещё не отправлено
}


public class ShopDb : DbContext
{
    public DbSet<Order> Orders => Set<Order>();
    public DbSet<OutboxMessage> Outbox => Set<OutboxMessage>();
    protected override void OnConfiguring(DbContextOptionsBuilder options) =>
        options.UseSqlServer("Server=.;Database=Shop;Trusted_Connection=True;TrustServerCertificate=True");
}
SQL: inbox и захват строк несколькими relay
-- Inbox: первичный ключ не даст обработать сообщение дважды
CREATE TABLE InboxMessages (
    MessageId   uniqueidentifier NOT NULL PRIMARY KEY,
    ProcessedAt datetime2        NOT NULL DEFAULT SYSUTCDATETIME()
);


BEGIN TRAN;
INSERT INTO InboxMessages (MessageId) VALUES (@MessageId); -- дубль: ошибка 2627, откат и ack
UPDATE Accounts SET Balance = Balance - @Amount WHERE Id = @AccountId;
COMMIT;


-- Relay в нескольких экземплярах (внутри транзакции): каждый берёт свои строки, занятые пропускает
SELECT TOP (100) Id, MessageId, AggregateId, Payload
FROM OutboxMessages WITH (UPDLOCK, READPAST, ROWLOCK)
WHERE SentAt IS NULL
ORDER BY Id;

READPAST позволяет relay работать параллельно, но порядок между экземплярами уже не гарантирован.

Источники
Senior+

Как выполнить бизнес-операцию через несколько сервисов без распределённой транзакции?

Заголовок раздела «Как выполнить бизнес-операцию через несколько сервисов без распределённой транзакции?»
Ответ

Что это

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

Как это работает

  1. Сервис заказов создаёт заказ со статусом Pending.
  2. Сервис оплаты списывает деньги.
  3. Сервис склада резервирует товар.
  4. Сервис заказов переводит заказ в Confirmed.

Что если шаг не удался

Товара нет — шаг 3 не выполнен. Тогда в обратном порядке идут компенсации: оплата возвращает деньги, заказ получает статус Cancelled.

Частая ошибка

Считать компенсацию откатом, который «стирает» шаг. Списание уже было, клиент мог получить SMS — компенсация это новое бизнес-действие со своими последствиями.

Как устроено

Признак Хореография Оркестрация
Кто ведёт процесс никто: сервисы реагируют на события оркестратор рассылает команды
Где видно состояние размазано по сервисам в одной записи саги
Подходит 2–3 шага, простой поток много шагов, ветвления, таймауты

Заказ с оркестратором

  1. Оркестратор создаёт запись саги в состоянии AwaitingPayment и отправляет команду ChargePayment.
  2. Получив PaymentCompleted, переходит в AwaitingStock и отправляет ReserveStock.
  3. Получив StockReserved, подтверждает заказ — сага завершена.
  4. Получив StockUnavailable, отправляет компенсацию RefundPayment, а после возврата отменяет заказ.
  5. Ответа нет дольше таймаута — повторяет команду или запускает компенсации.

Ловушки

  • Сообщения приходят at-least-once: шаги и компенсации должны быть идемпотентными.
  • Компенсация тоже может упасть — её повторяют до успеха.
  • Смену состояния саги и отправку команды связывают через outbox, иначе сага зависнет между шагами.

Правило

У каждого шага до точки невозврата есть компенсация, у каждого ожидания — таймаут.

Почему не 2PC

  • Двухфазный коммит держит блокировки у всех участников, пока координатор не объявит исход. Упал координатор между фазами — ресурсы остаются заблокированными в состоянии in doubt.
  • Kafka, RabbitMQ и HTTP-API в MSDTC не участвуют, а распределённые транзакции в .NET 7+ работают только на Windows.
  • Каждый участник становится обязательным для коммита, и доступность операции падает.

Под капотом

  • Состояние саги — запись в базе оркестратора: id, текущий шаг, данные, срок ожидания. Её обновляют с оптимистической блокировкой: события одной саги могут прийти параллельно.
  • Таймауты — запланированные сообщения или опрос записей с истёкшим сроком.
  • Шаги делятся на компенсируемые, точку невозврата и повторяемые после неё. Необратимое, например отгрузку, ставят после всех проверок.

Компромиссы

Изоляции нет: другой запрос увидит списанные деньги у ещё не подтверждённого заказа. Azure Architecture Center предлагает контрмеры: семантическую блокировку (статус Pending, который учитывают другие операции), коммутативные обновления, перечитывание значения перед изменением.

Как проверить в проде

  • Саги, застрявшие в одном состоянии дольше SLA, — главная метрика и алерт.
  • Корреляционный id саги в логах и трейсах связывает шаги разных сервисов.

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

Оркестратор саги как конечный автомат
var saga = new OrderSaga(Guid.NewGuid());
Print(saga.Start());                        // ChargePayment
Print(saga.Handle(new PaymentCompleted())); // ReserveStock
Print(saga.Handle(new StockUnavailable())); // RefundPayment — компенсация
Print(saga.Handle(new StockUnavailable())); // (пусто) — дубль игнорируется
Print(saga.Handle(new PaymentRefunded()));  // CancelOrder
Console.WriteLine(saga.State);              // Cancelled


static void Print(object[] commands) =>
    Console.WriteLine(string.Join(", ", commands.Select(c => c.GetType().Name)));


public enum SagaState { New, AwaitingPayment, AwaitingStock, Compensating, Completed, Cancelled }


public record PaymentCompleted; public record PaymentFailed; public record PaymentRefunded;
public record StockReserved; public record StockUnavailable;
public record ChargePayment(Guid OrderId); public record RefundPayment(Guid OrderId);
public record ReserveStock(Guid OrderId); public record ConfirmOrder(Guid OrderId);
public record CancelOrder(Guid OrderId);


public class OrderSaga(Guid orderId)
{
    public SagaState State { get; private set; } = SagaState.New;


    public object[] Start() => Go(SagaState.AwaitingPayment, new ChargePayment(orderId));


    // (текущее состояние, событие) → новое состояние и команды участникам
    public object[] Handle(object @event) => (State, @event) switch
    {
        (SagaState.AwaitingPayment, PaymentCompleted) => Go(SagaState.AwaitingStock, new ReserveStock(orderId)),
        (SagaState.AwaitingPayment, PaymentFailed) => Go(SagaState.Cancelled, new CancelOrder(orderId)),
        (SagaState.AwaitingStock, StockReserved) => Go(SagaState.Completed, new ConfirmOrder(orderId)),
        (SagaState.AwaitingStock, StockUnavailable) => Go(SagaState.Compensating, new RefundPayment(orderId)),
        (SagaState.Compensating, PaymentRefunded) => Go(SagaState.Cancelled, new CancelOrder(orderId)),
        _ => [] // дубль или устаревшее событие: ничего не делаем
    };


    private object[] Go(SagaState next, params object[] commands) { State = next; return commands; }
}

В реальной системе State и данные саги хранятся в базе, команды уходят через outbox, а таймауты приходят как отложенные сообщения.

Источники