MassTransit对接RabbitMQ时如何避免异常消息进入_error队列

需求实现可行性 #

你的需求完全可以通过MassTransit配置实现，只需要修改默认的异常消息直接路由到_error队列的逻辑，让抛出未处理异常的消息重新入队到原队列即可，参考配置代码如下：

cfg.ReceiveEndpoint("你的队列名称", e =>
{
    e.PrefetchCount = 1; // 保留你现有预取1的配置
    e.Consumer<你自定义的消费者类>();
    
    // 关闭默认错误消息直接转死信的逻辑
    e.DiscardFaultedMessages = false;
    // 异常触发时直接将消息重新入队
    e.OnError(context =>
    {
        context.Requeue();
        return Task.CompletedTask;
    });
    
    // 可按需配置内置重试策略，不需要就配置为不重试
    e.UseRetry(r => r.None());
});

配置完成后，消费者抛出未处理异常时的消息流转逻辑就和消费者崩溃未返回ACK的逻辑完全一致，消息会回到原队列，被其他在线消费者或者后续恢复的消费者拉取处理。

消息处理策略合理性分析 #

这个策略是否合理要结合你的业务场景判断：

• 适用且合理的场景：你的消息本身不存在业务合法性问题，故障均为偶发的临时故障（比如消费者节点临时资源不足、依赖的第三方服务偶发超时、数据库短时间不可用等），且所有消费者逻辑已经实现了幂等处理，这种场景下该策略可以避免临时故障导致的消息进入_error队列需要人工干预的问题，提升可用性。
• 不适用且不合理的场景：存在非法消息（比如格式错误、业务参数不符合要求，所有消费者处理都会抛出异常）、消费者逻辑未实现幂等，这种场景下该策略会导致非法消息在队列中无限循环，占用队列资源，甚至出现重复消费导致的业务数据错误。

异常处理方案选择建议 #

绝大多数场景下更推荐在消费者内部做异常兜底处理，尽量避免直接抛出未处理异常：

• 你可以对异常做分类处理：临时异常（网络抖动、依赖服务超时等）可以在消费者内部做有限次数的重试，重试失败再将消息重新入队；永久异常（消息格式错误、业务数据不存在等）直接记录告警日志，将消息转存到专用的死信存储，直接ACK掉消息避免无效循环。
• 内部兜底处理的灵活性远高于全局中间件配置，可以针对不同的业务消息、不同的异常类型做差异化的处理逻辑，也能减少中间件层的不必要开销。
• 无论选择哪种异常处理逻辑，竞争消费者场景下都必须保证消费者逻辑的幂等性，避免重复消费导致的业务问题。

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