C#如何实现MongoDB与Kafka操作的事务一致性

你当前的实现从根本上无法达到预期效果，核心原因有两个：

1. .NET自带的TransactionScope默认依赖MSDTC分布式事务协调器，Confluent.Kafka官方客户端从未实现MSDTC资源管理器适配，Kafka的生产逻辑根本不会参与事务的提交/回滚流程，哪怕Kafka发送抛异常，MongoDB的写入也不会自动回滚。
2. 你写的示例代码本身就有语法错误：using(var transactionScope = new TransactionScope());末尾多了分号，会导致using的作用域直接在分号处结束，后面大括号里的所有代码根本不在事务范围内；另外ProduceAsync是异步方法，你没有加await，方法抛出的异常不会被当前作用域捕获，完全无法触发事务回滚。

异构存储之间不存在开箱即用的强一致分布式事务，生产环境最成熟可靠的方案是事务性发件箱模式（Transactional Outbox），不需要依赖任何分布式事务协调器，就能保证MongoDB写入和Kafka消息投递的最终一致性。

整个模式的核心逻辑是把跨两个系统的原子操作，拆成单数据库内的原子操作+可靠重试投递：

• 所有业务写入和待发送消息的持久化，放在同一个MongoDB原生事务中完成，保证两者要么同时成功要么同时失败
• 独立的投递组件负责把持久化的消息发送到Kafka，发送成功后标记消息状态，失败则持续重试
• 配合幂等配置避免重复投递带来的业务问题

前置要求 #

• MongoDB必须部署为副本集或分片集群，单节点实例不支持事务
• Kafka生产者开启幂等配置，避免重试导致消息重复写入Kafka
• 消费者端基于消息唯一ID做幂等校验，兼容极端场景下的重复投递

代码实现 #

首先定义Outbox消息存储实体：

public enum OutboxStatus
{
    Pending,
    Sent
}

public class OutboxMessage
{
    public Guid Id { get; set; }
    public string Topic { get; set; }
    public string Key { get; set; }
    public string Payload { get; set; }
    public OutboxStatus Status { get; set; }
    public DateTime CreatedAt { get; set; }
    public DateTime? SentAt { get; set; }
}

业务写入逻辑，替换你原来的TransactionScope实现：

var client = new MongoClient(MongoClientSettings.FromConnectionString(mongoConnStr));
var _database = client.GetDatabase(mongoUrl.DatabaseName);
var drinkCollection = _database.GetCollection<Drinks>("drinks");
var orderCollection = _database.GetCollection<Orders>("orders");
var outboxCollection = _database.GetCollection<OutboxMessage>("outbox_messages");

// 使用Mongo原生会话事务，不要用TransactionScope
using var session = await client.StartSessionAsync();
session.StartTransaction();
try
{
    // 写入业务数据，所有操作必须传入session参数
    await drinkCollection.InsertOneAsync(session, new Drinks{ /* 你的饮品数据 */ });
    var newOrder = new Orders { /* 你的订单数据 */ };
    await orderCollection.InsertOneAsync(session, newOrder);

    // 不直接发送Kafka消息，先把消息序列化后和业务数据存在同一个库
    var pendingMsg = new OutboxMessage
    {
        Id = Guid.NewGuid(), // 全局唯一ID，用于幂等校验
        Topic = "commands",
        Key = newOrder.Id.ToString(), // 对应你原来的orderID作为消息键
        Payload = JsonSerializer.Serialize(new OrderCommand { Waiter = "John", Price = 3.35m }),
        Status = OutboxStatus.Pending,
        CreatedAt = DateTime.UtcNow
    };
    await outboxCollection.InsertOneAsync(session, pendingMsg);

    // 提交事务，到这一步业务数据和待发消息就同时持久化完成，不会出现部分成功
    await session.CommitTransactionAsync();
}
catch (Exception)
{
    // 任何异常直接回滚事务，所有Mongo侧写入全部撤销
    await session.AbortTransactionAsync();
    throw;
}

后台消息投递服务（可以用.NET的BackgroundService实现常驻运行）：

// Kafka生产者初始化时记得开启幂等配置
var producerConfig = new ProducerConfig
{
    BootstrapServers = kafkaConnStr,
    EnableIdempotence = true, // 开启幂等，避免重试导致Kafka侧重复消息
    Acks = Acks.All
};
using var _kafkaProducer = new ProducerBuilder<string, string>(producerConfig).Build();

// 常驻轮询待发送消息
while (!stoppingToken.IsCancellationRequested)
{
    // 建议给outbox_messages集合的Status、CreatedAt字段加索引提升查询效率
    var pendingMessages = await outboxCollection.Find(m => m.Status == OutboxStatus.Pending)
        .Limit(100)
        .ToListAsync(stoppingToken);
    
    foreach (var msg in pendingMessages)
    {
        try
        {
            var kafkaMsg = new Message<string, string>
            {
                Key = msg.Key,
                Value = msg.Payload
            };
            // 把消息ID放到Header，消费者取到后做幂等判断
            kafkaMsg.Headers.Add("message-id", msg.Id.ToByteArray());
            
            await _kafkaProducer.ProduceAsync(msg.Topic, kafkaMsg, stoppingToken);
            
            // 发送成功，更新消息状态为已发送
            await outboxCollection.UpdateOneAsync(
                Builders<OutboxMessage>.Filter.Eq(x => x.Id, msg.Id),
                Builders<OutboxMessage>.Update.Set(x => x.Status, OutboxStatus.Sent)
                    .Set(x => x.SentAt, DateTime.UtcNow),
                cancellationToken: stoppingToken);
        }
        catch (ProduceException<string,string> ex)
        {
            // 记录日志，不删除消息，下一轮循环自动重试
            _logger.LogError(ex, "Kafka消息投递失败，消息ID：{MsgId}", msg.Id);
        }
    }

    // 没有待发消息时等待500ms再轮询，避免打满数据库
    if (!pendingMessages.Any()) await Task.Delay(500, stoppingToken);
}

如果对消息投递延迟要求很高，可以不用定时轮询，改用MongoDB Change Stream监听outbox_messages集合的插入事件，有新消息时立刻触发投递，延迟可以降到毫秒级，核心逻辑和上面的实现完全一致。

不要尝试用两阶段提交（2PC）、先写库再发消息失败就回滚库这类方案：前者实现复杂度极高、故障恢复逻辑坑多，性能极差；后者在Kafka请求超时场景下无法判断消息是否已经投递成功，会出现消息发出去了但库被回滚的不一致问题，没有兜底手段。

内容的提问来源于stack exchange，提问作者101000101