使用MassTransit与Outbox Pattern时消息无法投递至RabbitMQ

问题描述 #

使用MassTransit结合Outbox Pattern时，消息成功写入数据库OutboxMessages表，但无法投递到RabbitMQ。当RabbitMQ宕机时Outbox能正常暂存消息，但RabbitMQ恢复后，消息仍无法被投递。

依赖包配置 #

<PackageReference Include="MassTransit" Version="8.0.15" />
<PackageReference Include="MassTransit.EntityFrameworkCore" Version="8.0.14" />
<PackageReference Include="MassTransit.RabbitMQ" Version="8.0.14" />

Program.cs配置 #

private static void AddMessagingConfiguration(IServiceCollection services, WebApplicationBuilder builder)
{
    services.AddMassTransit(busConfigurator =>
    {
        busConfigurator.AddConsumer<GtsDocumentConsumer>();
        busConfigurator.AddEntityFrameworkOutbox<CompanyGoodsDbContext>(outboxConfigurator =>
        {
            outboxConfigurator.UseSqlServer();
            outboxConfigurator.UseBusOutbox();
            outboxConfigurator.QueryMessageLimit = 10;
            outboxConfigurator.QueryDelay = TimeSpan.FromSeconds(5);
            outboxConfigurator.QueryTimeout = TimeSpan.FromSeconds(2);
        });

        busConfigurator.UsingRabbitMq((busContext, transportConfigurator) =>
        {              
            transportConfigurator.Host(builder.Configuration["EventBusSettings:HostAddress"], hostConfiguration =>
            {
                hostConfiguration.Username(builder.Configuration["EventBusSettings:Username"]);
                hostConfiguration.Password(builder.Configuration["EventBusSettings:Password"]);
                
            });

            transportConfigurator.AutoStart = true;
            transportConfigurator.UseMessageRetry(r => r.Immediate(5));
            transportConfigurator.UseScheduledRedelivery(r =>
                r.Intervals(TimeSpan.FromSeconds(5), TimeSpan.FromMinutes(15), TimeSpan.FromMinutes(30)));
            transportConfigurator.ConfigureEndpoints(busContext);
        });
    });
}

appsettings.json配置 #

"EventBusSettings": {
    "HostAddress": "sgms-broker",
    "Username": "guest",
    "Password": "guest"
}

注：应用已docker化，运行正常，sgms-broker是指向本地message broker的主机名。

控制器代码 #

private readonly IPublishEndpoint _publish;

public GtsDocumentsController(IPublishEndpoint publish)
{
    _publish = publish;
}

[HttpPut("add-seen")]
public async Task<IActionResult> AddSeenHistory([FromBody] AddSeenHistoryCommand command)
{
    await _publish.Publish(command);
    await documentCommandFacade.AddSeenHistory(command);
    return Ok();
}

排查与解决建议 #

• 统一MassTransit版本：当前依赖包中主包MassTransit为8.0.15，而MassTransit.EntityFrameworkCore和MassTransit.RabbitMQ为8.0.14，版本不一致可能引发兼容性问题，建议将所有MassTransit相关包版本统一。
• 验证数据库上下文与表结构：确认CompanyGoodsDbContext已正确配置OutboxMessages表的映射，且已执行数据库迁移确保表结构符合要求。
• 检查RabbitMQ连通性：在应用容器内测试与sgms-broker的5672端口连通性（如telnet sgms-broker 5672），同时查看应用日志是否存在RabbitMQ连接失败的错误信息。
• 确认总线与Outbox启动状态：检查应用启动日志，确认总线是否成功初始化，AutoStart = true是否生效，Bus Outbox的轮询服务是否正常启动。
• 确保事务一致性：控制器中_publish.Publish与documentCommandFacade.AddSeenHistory需处于同一数据库事务中，只有业务操作成功提交后，Outbox消息才会被标记为待投递。可通过将两者包裹在数据库事务中实现。
• 开启调试日志排查：启用MassTransit的调试级别日志，查看Outbox轮询消息、尝试投递过程中的详细日志，定位是否存在数据库查询超时、权限不足或其他异常。

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