使用MassTransit+RabbitMQ时消息进入skipped队列及交换命名异常

场景说明 #

有两个应用分别负责消息发布与订阅，采用多总线模式（自定义IEmployeeSvcBus接口），相关配置代码如下：

发布端总线注册代码 #

context.Services.AddMassTransit<IEmployeeSvcBus>(x =>
{                
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost:5672");

        cfg.Message<ICreateEmployeeEto>(m => m.SetEntityName("CreateEmployee"));
        cfg.Message<IUpdateEmployeeEto>(m => m.SetEntityName("UpdateEmployee"));
        
        cfg.Publish<ICreateEmployeeEto>(y =>
        {
            y.ExchangeType = ExchangeType.FanOut.ToString().ToLower();
            y.AutoDelete = false;
            y.Durable = true;
        });

        cfg.Publish<IUpdateEmployeeEto>(y =>
        {
            y.ExchangeType = ExchangeType.FanOut.ToString().ToLower();
            y.AutoDelete = false;
            y.Durable = true;
        });                    
    });
});

自定义总线接口 #

public interface IEmployeeSvcBus: IBus
{
}

消费端总线配置代码 #

context.Services.AddMassTransit<IEmployeeSvcBus>(x =>
{
    x.AddConsumer<CreateEmployeeConsumer>();
    x.AddConsumer<UpdateEmployeeConsumer>();

    x.UsingRabbitMq((context, busCfg) =>
    {
        busCfg.Host("rabbitmq://localhost:5672");
        
        busCfg.ReceiveEndpoint("employees.create_dev", ep =>
        {
            ep.Bind("CreateEmployee");
            ep.ConfigureConsumeTopology = false;
            ep.Durable = true;
            ep.Lazy = true;
            ep.Consumer<CreateEmployeeConsumer>();
        });

        busCfg.ReceiveEndpoint("employees.update_dev", ep =>
        {
            ep.Bind("UpdateEmployee");
            ep.ConfigureConsumeTopology = false;
            ep.Durable = true;
            ep.Lazy = true;
            ep.Consumer<UpdateEmployeeConsumer>();
        });
    });
});

消费者类定义 #

public class CreateEmployeeConsumer: IConsumer<ICreateEmployeeEto>
{
    private readonly IBus _localBus;
    private readonly ImySvc _mySvc;

    public CreateEmployeeConsumer(
        IBus localBus,
        ImySvc mySvc)
    {
        _localBus = localBus;
        _mySvc= mySvc;
    }

    public CreateEmployeeConsumer()
    {
        //For Bus Registration
    }

    public async Task Consume(ConsumeContext<ICreateEmployeeEto> context)
    {
        //Some operations.
    }
}

遇到的问题 #

1. 交换器名称不符合预期：原本期望ICreateEmployeeEto和IUpdateEmployeeEto消息分别发布到CreateEmployee、UpdateEmployee交换器，但实际生成了带命名空间的交换器<namespace>:ICreateEmployeeEto和<namespace>:IUpdateEmployeeEto，目前通过RabbitMQ手动绑定临时解决。
2. 消息进入skipped队列：发布CreateEmployeeEto消息后，消息进入employees.create_dev_skipped队列，尝试注释依赖注入、移除绑定等操作后问题依旧。

问题原因及解决方法 #

问题1：交换器名称不符合预期 #

原因：消息实体名称的配置位置错误。当前代码在UsingRabbitMq的上下文cfg中配置Message<T>，但多总线模式下，该配置应放在AddMassTransit的总线注册配置器x上，否则MassTransit会使用默认的「命名空间+类型名」作为交换器名称。

解决方法：调整发布端的配置，将Message<T>的配置移到x的层级下：

context.Services.AddMassTransit<IEmployeeSvcBus>(x =>
{
    // 将消息实体名称配置移到这里
    x.Message<ICreateEmployeeEto>(m => m.SetEntityName("CreateEmployee"));
    x.Message<IUpdateEmployeeEto>(m => m.SetEntityName("UpdateEmployee"));

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost:5672");
        
        cfg.Publish<ICreateEmployeeEto>(y =>
        {
            y.ExchangeType = ExchangeType.FanOut.ToString().ToLower();
            y.AutoDelete = false;
            y.Durable = true;
        });

        cfg.Publish<IUpdateEmployeeEto>(y =>
        {
            y.ExchangeType = ExchangeType.FanOut.ToString().ToLower();
            y.AutoDelete = false;
            y.Durable = true;
        });                    
    });
});

问题2：消息进入skipped队列 #

原因：本质是消息类型不匹配导致无法被消费者处理。由于问题1中发布端生成了带命名空间的交换器，手动绑定后消息的类型标识（<namespace>:ICreateEmployeeEto）与消费端消费者订阅的ICreateEmployeeEto不匹配，MassTransit无法将消息路由到正确的消费者，因此放入skipped队列。

解决方法：

1. 先修复问题1，确保发布端正确将消息发送到CreateEmployee交换器。
2. 在消费端的AddMassTransit配置中，添加与发布端一致的消息实体名称配置，保证类型标识匹配：

context.Services.AddMassTransit<IEmployeeSvcBus>(x =>
{
    // 添加消息实体名称配置
    x.Message<ICreateEmployeeEto>(m => m.SetEntityName("CreateEmployee"));
    x.Message<IUpdateEmployeeEto>(m => m.SetEntityName("UpdateEmployee"));

    x.AddConsumer<CreateEmployeeConsumer>();
    x.AddConsumer<UpdateEmployeeConsumer>();

    x.UsingRabbitMq((context, busCfg) =>
    {
        busCfg.Host("rabbitmq://localhost:5672");
        
        busCfg.ReceiveEndpoint("employees.create_dev", ep =>
        {
            ep.Bind("CreateEmployee");
            ep.ConfigureConsumeTopology = false;
            ep.Durable = true;
            ep.Lazy = true;
            ep.Consumer<CreateEmployeeConsumer>();
        });

        busCfg.ReceiveEndpoint("employees.update_dev", ep =>
        {
            ep.Bind("UpdateEmployee");
            ep.ConfigureConsumeTopology = false;
            ep.Durable = true;
            ep.Lazy = true;
            ep.Consumer<UpdateEmployeeConsumer>();
        });
    });
});

3. 移除消费者的无参构造函数：MassTransit会通过依赖注入自动解析构造函数参数，无需手动添加无参构造，避免可能的DI冲突。

排查建议 #

• 查看RabbitMQ管理界面，检查交换器的绑定关系、消息的type属性是否与消费者订阅的类型一致。
• 启用MassTransit的日志（如Serilog、NLog），查看消息处理时的详细日志，尤其是反序列化、类型匹配相关的报错信息。
• 测试时可先简化配置：去掉多总线、自定义交换器配置，验证基础的发布/订阅是否正常，再逐步添加自定义配置。

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