如何在MassTransit与Entity Framework事务中通过ISendEndpointProvider发消息

核心问题答复 #

ISendEndpointProvider 完全支持事务操作，结合你当前的技术栈版本，有两种成熟可行的方案可以保证数据库更新与消息发送的原子性。

方案1：基于TransactionScope的分布式事务方案 #

实现逻辑 #

通过.NET自带的TransactionScope将数据库更新、消息发送两个操作纳入同一个分布式事务，只有两个操作全部执行成功才会提交事务，任意一步失败都会触发整体回滚，避免消息丢失或重复发送。

代码修改 #

public sealed class ScheduleMessageJob : IJob
{
    private readonly IServiceProvider _serviceProvider;

    public ScheduleMessageJob(IServiceProvider serviceProvider) =>
        _serviceProvider = serviceProvider;

    public async Task Execute(IJobExecutionContext context)
    {
        using var scope = _serviceProvider.CreateScope();
        var scopedServiceProvider = scope.ServiceProvider;
        // 必须开启异步事务流支持，否则异步上下文内事务会失效
        using var transactionScope = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled);
        
        await UpdateScheduledMessage(scopedServiceProvider);
        await Send(scopedServiceProvider);
        
        // 两步操作全部成功后再提交事务
        transactionScope.Complete();
    }

    private static async Task UpdateScheduledMessage(IServiceProvider scopedServiceProvider)
    {
        var dbContext = scopedServiceProvider.GetRequiredService<IDbContext>();
        var scheduledMessage = await dbContext.Get<ScheduledMessage>(id: 1);
        scheduledMessage.IsQueued = true;
        dbContext.Update(scheduledMessage);
        // EF Core 5默认会自动登记到当前活动的TransactionScope，无需额外操作
        await dbContext.SaveChangesAsync();
    }

    public static async Task Send(IServiceProvider scopedServiceProvider)
    {
        var endpoint = await GetEndpoint(scopedServiceProvider);
        var triggerExecuted = new TriggerExecuted("Some data");
        // MassTransit 7.2.3默认会自动将发送操作登记到当前活动事务
        await endpoint.Send(triggerExecuted);
    }

    private static async Task<ISendEndpoint> GetEndpoint(IServiceProvider serviceProvider) =>
        await serviceProvider
        .GetRequiredService<ISendEndpointProvider>()
        .GetSendEndpoint(new Uri("queue:SomeQueue"));
}

注意事项 #

• 该方案依赖分布式事务协调器（Windows环境为MSDTC，Linux环境为.NET Core内置分布式事务支持），需要部署环境开启对应服务
• 确保RabbitMQ配置未禁用事务功能，MassTransit 7.x版本默认已开启事务支持无需额外配置

方案2：基于MassTransit EF发件箱的本地事务方案（更推荐） #

实现逻辑 #

使用MassTransit内置的Entity Framework Outbox功能，将消息发送操作与业务数据更新共享同一个数据库本地事务，完全不需要依赖分布式事务。
消息会先和业务数据一起写入数据库的Outbox表，事务提交成功后，MassTransit后台服务会自动将Outbox中的消息投递到RabbitMQ，投递成功后会删除Outbox记录。即使投递过程中应用崩溃，下次启动时会自动扫描未投递的Outbox消息继续投递，不会丢失也不会重复调度。

配置步骤 #

1. 服务注册时添加EF Outbox配置，关联你项目中的DbContext
2. Job代码无需调整事务包裹逻辑，仅需保证数据库更新与消息发送在同一个DbContext的事务范围内即可

方案优势 #

• 不依赖分布式事务，跨平台兼容性更好，性能更高
• 完全兼容你现有的故障恢复逻辑，不需要额外处理重复发送问题

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