MassTransit无法消费Fault消息，消息仅进入_error队列请求排查

我正尝试编写错误消息处理器，但消息仅进入带有_error前缀的队列后就无后续动作，请问我哪里配置出错了？

消费者代码 #

public class CustomerCreatedEventConsumer : 
    IConsumer<CustomerCreatedArgs>
{
    public Task Consume(ConsumeContext<CustomerCreatedArgs> context)
    {
        throw new Exception("TEST EXCEPTION!!!!");
        return Task.CompletedTask;
    }
}

public class DashboardFaultConsumer :
    IConsumer<Fault<CustomerCreatedArgs>>
{
    public Task Consume(ConsumeContext<Fault<CustomerCreatedArgs>> context)
    {
        /// !!!
        ///  这段代码应该执行，但实际没有触发
        /// !!!
        return Task.CompletedTask;
    }
}

配置代码 #

public static class ConfigureServicesMassTransit
{
    public static void ConfigureMassTransitServices(this IServiceCollection services, IConfiguration configuration)
    {
        var massTransitSection = configuration.GetSection("MassTransit");
        var url = massTransitSection.GetValue<string>("Url");
        var host = massTransitSection.GetValue<string>("Host");
        var userName = massTransitSection.GetValue<string>("UserName");
        var password = massTransitSection.GetValue<string>("Password");

        if (massTransitSection == null || url == null || host == null)
        {
            throw new ArgumentNullException("Section 'MassTransit' configuration settings are not found in appsettings.json");
        }

        services.AddMassTransit(x =>
        {
            AddConsumers(x);
            x.UsingRabbitMq((context, cfg) =>
            {
                cfg.Host($"{url}", configurator =>
                {
                    configurator.Username(userName);
                    configurator.Password(password);
                });

                cfg.ConfigureEndpoints(context);
            });
        });
    }

    private static void AddConsumers(IServiceCollectionBusConfigurator x)
    {
        x.AddConsumer<CustomerCreatedEventConsumer>();
        x.AddConsumer<DashboardFaultConsumer>();
    }
}

核心问题是默认情况下MassTransit不会自动发布Fault<T>事件，消息消费失败后直接进入死信队列（_error后缀队列），并未触发Fault消费者的订阅逻辑。

你需要做以下两处调整：

1. 启用Fault发布并配置重试策略 #

在RabbitMQ配置中，显式启用PublishFaults并配置重试策略，确保消费失败时发布Fault事件：

x.UsingRabbitMq((context, cfg) =>
{
    cfg.Host($"{url}", configurator =>
    {
        configurator.Username(userName);
        configurator.Password(password);
    });

    // 为CustomerCreatedEventConsumer配置接收端点及重试、Fault发布
    cfg.ReceiveEndpoint("customer-created-event", e =>
    {
        e.ConfigureConsumer<CustomerCreatedEventConsumer>(context);
        e.UseMessageRetry(r => r.Interval(2, 1000)); // 重试2次，每次间隔1秒，可按需调整
        e.PublishFaults = true; // 启用Fault事件发布
    });

    cfg.ConfigureEndpoints(context);
});

2. 确认Fault消费者的端点配置 #

虽然ConfigureEndpoints会自动为DashboardFaultConsumer创建端点，但也可以显式配置以确保绑定正确：

// 在UsingRabbitMq代码块内添加
cfg.ReceiveEndpoint("dashboard-fault-handler", e =>
{
    e.ConfigureConsumer<DashboardFaultConsumer>(context);
});

补充说明 #

• _error后缀队列是MassTransit的死信队列，消息重试失败后会被转移到这里，属于正常行为，但该队列的消息不会触发Fault消费者。
• 只有启用PublishFaults并配置重试后，消费失败时才会发布Fault<CustomerCreatedArgs>事件，你的DashboardFaultConsumer才能接收到事件并执行逻辑。

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