MassTransit连接RabbitMQ运行中断时消息丢失，如何在发布前校验连接状态

你当前使用的IsStarted属性仅标识发布器是否完成了启动初始化流程，不会同步后续运行过程中的连接状态，所以无法作为发布前的校验依据。运行中连接断开时消息无异常丢失的核心原因是默认未开启RabbitMQ发布者确认机制，客户端不会等待Broker的投递回执，连接异常时消息直接在传输层丢弃不会通知上层。

方案1：开启发布者确认机制，直接捕获发布异常（最推荐） #

MassTransit原生支持RabbitMQ发布者确认，开启后Publish方法会等待Broker返回投递成功回执后才完成，只要消息未成功投递就会抛出对应异常，你直接捕获异常做降级处理即可，不需要提前判断状态。
配置方式：

services.AddMassTransit(x =>
{
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://你的RabbitMQ地址");
        // 开启发布确认
        cfg.PublisherConfirmation = true;
        // 可搭配设置确认超时时间，避免长时间阻塞
        cfg.PublisherConfirmationTimeout = TimeSpan.FromSeconds(3);
    });
});

修改后的发布逻辑：

try
{
    await _messagePublisher.Publish(attempt, _messageRouter.ObtainRoutingKey(attempt));
}
catch (BrokerUnreachableException)
{
    // Broker连接不可用，执行本地暂存、告警等降级逻辑
}
catch (MessageNotConfirmedException)
{
    // 消息未收到Broker确认，执行重试、暂存逻辑
}
catch (Exception ex)
{
    // 其他异常处理
}

方案2：发布前实时检查连接状态 #

如果确实需要提前判断状态避免无效调用，可以通过MassTransit的健康检查接口获取真实连接状态：

// 构造函数注入IBusHealthCheck
private readonly IBusHealthCheck _busHealthCheck;

// 发布前判断逻辑
var healthCheckResult = await _busHealthCheck.CheckHealthAsync(new HealthCheckContext(), CancellationToken.None);
if (healthCheckResult.Status == HealthStatus.Healthy)
{
    await _messagePublisher.Publish(attempt, _messageRouter.ObtainRoutingKey(attempt));
}
else
{
    // 执行对应降级处理
}

如果你使用的MassTransit版本不支持IBusHealthCheck，也可以直接读取传输层连接状态：

var rabbitMqBus = _messagePublisher as IRabbitMqBus;
bool isConnected = rabbitMqBus?
    .TransportProbe
    .Result?
    .All(transport => transport.Status == TransportStatus.Connected) ?? false;

if (isConnected)
{
    await _messagePublisher.Publish(attempt, _messageRouter.ObtainRoutingKey(attempt));
}
else
{
    // 执行对应降级处理
}

方案3：增加本地消息表做兜底 #

不管用上述哪种校验方式，都存在「状态检查后、发布前」的极短时间窗口内连接断开的极端情况，建议增加本地消息表做兜底：

• 发布消息前先将消息持久化到本地数据库/本地文件，标记为待发布
• 消息发布成功后更新状态为已发布
• 后台启动定时任务，定期扫描待发布的消息，等连接恢复后自动补发

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