EF Core+PostgreSQL下无需2PC实现Rebus与事务联动的方案问询

好问题！你遇到的这个情况其实是因为Rebus.TransactionScopes包的通用设计导致的——它不知道你EF和Rebus用的是同一个PostgreSQL实例，所以默认启用了2PC（两阶段提交）来保证跨资源的一致性，这就要求PostgreSQL开启prepared transactions，而你刚好没开，所以报错了。

下面给你两个无需2PC就能实现事务一致性的方案，都是针对你当前技术栈的：

方案1：让Rebus直接参与EF的数据库事务 #

既然EF和Rebus都用同一个PostgreSQL数据库，完全可以绕开TransactionScope，让Rebus直接绑定到EF的DbContext事务上。这样所有操作都在同一个本地数据库事务里，自然不需要2PC。

代码示例：

public override int SaveChanges()
{
    using (var dbTransaction = Database.BeginTransaction())
    {
        try
        {
            // 先提交EF的业务变更
            var result = base.SaveChanges();

            // 获取EF使用的Npgsql连接
            var npgsqlConnection = Database.GetDbConnection() as NpgsqlConnection;
            if (npgsqlConnection == null)
            {
                throw new InvalidOperationException("Expected Npgsql connection");
            }

            // 创建Rebus事务并绑定到当前数据库事务
            using (var rebusTx = _bus.Advanced.TransactionContext.Create())
            {
                rebusTx.Enlist(npgsqlConnection, dbTransaction);
                // 发送消息（如果用异步SaveChangesAsync，要改成await调用）
                _bus.Send("something happened").Wait();
                rebusTx.CompleteAsync().Wait();
            }

            // 提交整个数据库事务
            dbTransaction.Commit();
            return result;
        }
        catch
        {
            dbTransaction.Rollback();
            throw;
        }
    }
}

注意：如果你的应用采用异步编程模型，建议改成SaveChangesAsync的异步版本，避免用.Wait()阻塞线程，保证代码的非阻塞性。

这个方案的核心是利用Rebus.PostgreSQL传输对现有数据库事务的支持，让消息发送和EF变更共享同一个本地事务，要么都成功，要么都回滚。

方案2：本地消息表+后台Worker（你提到的可靠方案） #

这个方案是业界常用的“本地事务表”模式，完全依赖数据库的ACID特性，不需要任何分布式事务，可靠性拉满，而且解耦了业务操作和消息发送。

具体步骤：

1. 在你的EF上下文里添加一个OutgoingMessages表，用来暂存待发送的消息
2. 在SaveChanges时，把要发送的消息插入到这个表，和业务数据变更在同一个EF事务里提交
3. 写一个后台Worker服务，定期拉取未处理的消息，用Rebus发送，发送成功后标记为已处理

代码示例 #

首先定义消息实体：

public class OutgoingMessage
{
    public Guid Id { get; set; } = Guid.NewGuid();
    public string MessageBody { get; set; }
    public string MessageType { get; set; }
    public bool IsProcessed { get; set; } = false;
    public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
}

在DbContext里配置表：

protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.Entity<OutgoingMessage>(b =>
    {
        b.HasKey(x => x.Id);
        b.Property(x => x.MessageBody).IsRequired();
        b.Property(x => x.MessageType).IsRequired();
        // 给未处理消息加索引，提高查询效率
        b.HasIndex(x => x.IsProcessed);
    });
}

修改SaveChanges方法：

public override int SaveChanges()
{
    // 添加待发送消息到本地表
    Add(new OutgoingMessage
    {
        MessageBody = JsonSerializer.Serialize("something happened"),
        MessageType = typeof(string).FullName
    });

    // 提交业务数据和消息的本地事务
    return base.SaveChanges();
}

后台Worker服务（用ASP.NET Core的BackgroundService）：

public class OutgoingMessageWorker : BackgroundService
{
    private readonly IBus _bus;
    private readonly YourDbContext _dbContext;
    private readonly ILogger<OutgoingMessageWorker> _logger;

    public OutgoingMessageWorker(IBus bus, YourDbContext dbContext, ILogger<OutgoingMessageWorker> logger)
    {
        _bus = bus;
        _dbContext = dbContext;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Outgoing message worker started");

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                // 批量拉取未处理的消息（一次取10条，避免压力过大）
                var messages = await _dbContext.OutgoingMessages
                    .Where(x => !x.IsProcessed)
                    .Take(10)
                    .ToListAsync(stoppingToken);

                foreach (var message in messages)
                {
                    try
                    {
                        // 反序列化消息并发送
                        var messageType = Type.GetType(message.MessageType);
                        var messageObject = JsonSerializer.Deserialize(message.MessageBody, messageType);
                        await _bus.Send(messageObject);

                        // 标记为已处理
                        message.IsProcessed = true;
                        _logger.LogInformation($"Sent message {message.Id} successfully");
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, $"Failed to send message {message.Id}, will retry later");
                        // 这里可以加重试次数限制，或者直接留着下次继续尝试
                    }
                }

                // 提交更新
                await _dbContext.SaveChangesAsync(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing outgoing messages");
            }

            // 每隔5秒轮询一次，可以根据业务调整
            await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
        }

        _logger.LogInformation("Outgoing message worker stopped");
    }
}

最后记得把Worker注册到DI容器：

services.AddHostedService<OutgoingMessageWorker>();

这个方案的优势 #

• 完全不需要2PC，所有操作都是本地数据库事务，避免了PostgreSQL配置的麻烦
• 消息不会丢失：即使Rebus或者消息中间件挂了，消息存在数据库里，Worker恢复后会继续发送
• 解耦业务逻辑：业务操作不用等待消息发送完成，提高响应速度
• 容易监控和调试：可以直接查数据库表看消息状态

总结 #

• 如果想要最直接的同事务发送，选方案1，利用Rebus和EF共享本地事务，代码改动小
• 如果想要更可靠、解耦的架构，选方案2，这是生产环境中非常推荐的模式，尤其是对高可用要求高的场景

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