如何结合MassTransit Mediator与Transactional Outbox实现事务一致性？

核心配置与代码调整 #

MassTransit Mediator默认是内存内消息传递，要和Transactional Outbox结合，关键是让命令处理过程绑定数据库事务上下文，确保数据操作与消息发布原子性。

1. 注册时集成Mediator与Outbox
   在服务注册阶段，需为Mediator启用Transactional Outbox，并配置Outbox关联数据库（以EF Core为例）：

builder.Services.AddMassTransit(x =>
{
    x.AddMediator(m =>
    {
        m.AddConsumer<CreateOrderConsumer>();
        // 为Mediator消费者启用Outbox支持
        m.UseTransactionalOutbox();
    });

    x.UsingInMemory((context, cfg) =>
    {
        cfg.ConfigureEndpoints(context);
    });

    // 配置Outbox使用数据库
    x.AddEntityFrameworkOutbox<AppDbContext>(o =>
    {
        o.QueryDelay = TimeSpan.FromSeconds(10);
        o.UseSqlServer();
        o.UseBusOutbox();
    });
});

2. 命令处理中绑定事务
   在应用层消费者（命令处理类）中，确保数据库操作与消息发布共享同一事务：

public class CreateOrderConsumer : IConsumer<CreateOrderCommand>
{
    private readonly AppDbContext _dbContext;
    private readonly IPublishEndpoint _publishEndpoint;

    public CreateOrderConsumer(AppDbContext dbContext, IPublishEndpoint publishEndpoint)
    {
        _dbContext = dbContext;
        _publishEndpoint = publishEndpoint;
    }

    public async Task Consume(ConsumeContext<CreateOrderCommand> context)
    {
        var order = new Order { /* 实体赋值 */ };
        _dbContext.Orders.Add(order);

        // 发布事件，该操作会被Outbox拦截存入数据库
        await _publishEndpoint.Publish(new OrderCreatedEvent { OrderId = order.Id });

        // 提交事务，订单数据与Outbox消息将原子性持久化
        await _dbContext.SaveChangesAsync();
    }
}

3. 修正上下文差异问题
   控制器中直接注入IMediator发送命令，确保作用域内DbContext与Outbox上下文关联：

[ApiController]
[Route("orders")]
public class OrdersController : ControllerBase
{
    private readonly IMediator _mediator;

    public OrdersController(IMediator mediator)
    {
        _mediator = mediator;
    }

    [HttpPost]
    public async Task<IActionResult> CreateOrder(CreateOrderCommand command)
    {
        await _mediator.Send(command);
        return Ok();
    }
}

替代方案 #

如果上述集成方式不满足需求，可选择以下方案实现层间通信+事务一致性：

• 使用MassTransit常规总线：放弃Mediator，直接基于RabbitMQ/Kafka等消息总线开发，结合Transactional Outbox。命令从控制器发送至总线，消费者处理时绑定数据库事务，Outbox自动拦截消息，保障数据与消息的原子性，这是MassTransit官方推荐的标准模式。
• 自定义命令总线+Outbox拦截器：自行实现类似MediatR的命令总线，在命令处理管道中加入Outbox拦截逻辑。开启数据库事务后执行命令处理，将待发布消息存入Outbox表，事务提交后通过后台任务将消息发布至队列。
• EF Core事务拦截器：利用EF Core的SaveChangesInterceptor，在数据保存时自动收集领域事件并存入Outbox表，事务提交后通过后台服务批量发布消息。该方案无需依赖MassTransit Mediator，仅结合EF Core与MassTransit Outbox组件即可。

内容的提问来源于stack exchange，提问作者Oscar Javier Jiménez Mejía