使用MassTransit向RabbitMQ发消息为何进入__skipped队列？

我在.NET Core API中配置了MassTransit和RabbitMQ，代码如下：

services.AddMassTransit(masstransit =>
{
    masstransit.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(new Uri("rabbitmq://srv-docker-dev-01"), "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.ReceiveEndpoint("implantacao", e =>
        {
            e.ConfigureConsumeTopology = false;
            e.Bind<ClienteAlteradoEvento>();
        });
    });
});

还没创建消费者，RabbitMQ里就出现了__skipped队列，消息都被投递到这个队列，请问哪里操作错了？

补充问题：

• 消费者无法访问生产者的事件类；
• 需要两个队列消费该API的消息，是不是要配置另一个接收端点？
• 我原本以为每个微服务对应一个Exchange，但MassTransit为每个事件类型创建一个Exchange，这是否正确？

编辑补充：

• 微服务A：提供接收客户输入数据的端点，客户数据更新时需发布消息同步其他微服务，包含CustomerDataUpdatedEvent、CustomerAddressUpdatedEvent、CustomerContractsUpdatedEvent三种事件；
• 微服务B、C：需消费微服务A发布的客户更新消息，但无法访问A的事件类，得自行定义类，请问这种方式可行、合理吗？

1. __skipped队列出现的原因 #

你配置了ReceiveEndpoint并绑定了ClienteAlteradoEvento事件，但没有为这个端点注册对应的消费者。当消息被投递到该端点队列后，MassTransit找不到处理消息的消费者，就会把消息移到__skipped队列。即使你还没创建消费者，只要绑定了事件，RabbitMQ的路由规则会把消息送到这个端点队列，没消费者处理就触发skipped机制。

解决方法：如果暂时不需要这个接收端点，先删掉这段配置；如果是为后续消费者预留，要么先注册一个空的消费者占位，要么等消费者实现后再添加绑定配置。

2. 消费者自定义事件类的可行性 #

完全可行且合理，但要注意几个关键细节：

• 消息契约一致性：自定义的事件类必须和生产者的事件类保持字段名称、类型完全一致，包括嵌套对象的结构。MassTransit是基于消息的JSON结构来匹配的，只要结构一致，即使类名或命名空间不同，也能正确反序列化。
• 消息类型标识：如果生产者的事件类设置了MessageUrn或者使用了默认的类型标识（基于类的全名），消费者的自定义类需要通过[MessageUrn]特性指定和生产者相同的URN，或者在配置里手动设置消息类型映射，避免MassTransit识别为不同的消息类型。
• 版本兼容性：后续更新消息结构时，要保证新旧版本的兼容性（比如新增字段设为可空），防止消费者反序列化失败。

3. MassTransit的Exchange设计逻辑 #

MassTransit为每个事件类型创建Exchange是正确且符合消息队列最佳实践的：

• 这种设计遵循事件驱动架构的单一职责原则，每个事件对应一个Exchange，消费者可以按需绑定自己需要的事件Exchange，实现精准订阅。
• 如果你想让一个微服务对应一个Exchange，也可以手动配置：创建一个总Exchange，然后把所有事件Exchange绑定到这个总Exchange上，消费者只需要绑定总Exchange就能接收所有事件。但这种方式会让消费者收到不需要的消息，增加过滤成本，一般不推荐。

4. 多队列消费的配置方式 #

要让两个队列消费同一个API的消息，只需要配置两个独立的ReceiveEndpoint即可：

masstransit.UsingRabbitMq((context, cfg) =>
{
    // Host配置省略

    // 第一个接收端点（对应微服务B）
    cfg.ReceiveEndpoint("microservice-b-customer-queue", e =>
    {
        e.ConfigureConsumeTopology = false;
        e.Bind<CustomerDataUpdatedEvent>();
        e.Bind<CustomerAddressUpdatedEvent>();
        // 注册微服务B的消费者
        e.Consumer<CustomerDataUpdatedConsumerB>(context);
    });

    // 第二个接收端点（对应微服务C）
    cfg.ReceiveEndpoint("microservice-c-customer-queue", e =>
    {
        e.ConfigureConsumeTopology = false;
        e.Bind<CustomerContractsUpdatedEvent>();
        // 注册微服务C的消费者
        e.Consumer<CustomerContractsUpdatedConsumerC>(context);
    });
});

每个接收端点对应一个独立队列，各自绑定需要的事件Exchange，注册对应的消费者即可实现独立消费。

内容的提问来源于stack exchange，提问作者Leandro De Mello Fagundes