MassTransit 5.5.3与RabbitMQ微服务事件发布订阅失败解决问询

你的猜测完全正确！问题的核心就是消息契约不匹配——MassTransit是通过**完整的类型名称（包括命名空间）**来识别消息契约的，Publisher.Message和Consumer.Message在MassTransit眼里是两个完全不同的消息类型，所以发布的消息找不到对应的消费者，要么无去向，要么被路由到_skipped队列（MassTransit用来存放无法匹配消费者的消息）。

下面给你两种解决方案，优先推荐第一种（最规范、易维护）：

方案1：使用共享类库统一消息契约 #

把消息接口提取到一个独立的类库项目中，让发布端和消费端都引用这个类库，这样两边的消息类型就完全一致了。

步骤1：创建共享类库的消息接口

namespace Shared.Messages {
    public interface Message {
        string Text { get; set; }
    }
}

步骤2：修改发布端代码

引用共享类库后，替换掉原来的本地Message接口：

namespace Publisher {
    class Program {
        static async Task Main(string[] args) {
            var bus = Bus.Factory.CreateUsingRabbitMq(cfg => {
                IRabbitMqHost host = cfg.Host("host", "vhost", h => {
                    h.Username("xxx");
                    h.Password("yyy");
                });
            });

            bus.Start();
            Console.WriteLine("Publishing message...");
            await bus.Publish<Shared.Messages.Message>(new { Text = "Hello World" });
            Console.ReadKey();
            bus.Stop();
        }
    }
}

步骤3：修改消费端代码

同样引用共享类库，替换消息类型并给接收队列指定明确名称（避免自动生成队列名的潜在问题）：

namespace Consumer {
    class Program {
        static async Task Main(string[] args) {
            var bus = Bus.Factory.CreateUsingRabbitMq(cfg => {
                IRabbitMqHost host = cfg.Host("host", "vhost", h => {
                    h.Username("xxx");
                    h.Password("yyy");
                });

                // 给队列起一个明确的名字，方便管理
                cfg.ReceiveEndpoint(host, "message-consumer-queue", e => {
                    e.Consumer<MbConsumer>();
                });
            });

            bus.Start();
            Console.WriteLine("Consumer is running. Waiting for messages...");
            Console.ReadKey();
            bus.Stop();
        }
    }

    public class MbConsumer : IConsumer<Shared.Messages.Message> {
        public async Task Consume(ConsumeContext<Shared.Messages.Message> context) {
            await Console.Out.WriteLineAsync($"Received message: {context.Message.Text}");
        }
    }
}

这样修改后，发布端和消费端使用的是同一个Shared.Messages.Message类型，MassTransit会自动绑定对应的交换器和队列，消息就能被正确消费了。

方案2：显式配置消息契约映射（不推荐，仅用于无法使用共享类库的场景） #

如果因为某些原因不能用共享类库，可以通过显式配置让两个不同的类型映射到同一个消息契约。在发布端和消费端的Bus配置中添加以下代码：

发布端配置

var bus = Bus.Factory.CreateUsingRabbitMq(cfg => {
    IRabbitMqHost host = cfg.Host("host", "vhost", h => {
        h.Username("xxx");
        h.Password("yyy");
    });

    // 给本地Message类型指定统一的契约名称
    cfg.Message<Message>(x => {
        x.SetEntityName("shared.message");
        x.SetMessageUrn("urn:message:shared:message");
    });

    // 确保两边使用相同的序列化器（比如Json）
    cfg.SetMessageSerializer(new JsonMessageSerializer());
});

消费端配置

var bus = Bus.Factory.CreateUsingRabbitMq(cfg => {
    IRabbitMqHost host = cfg.Host("host", "vhost", h => {
        h.Username("xxx");
        h.Password("yyy");
    });

    cfg.ReceiveEndpoint(host, "message-consumer-queue", e => {
        e.Consumer<MbConsumer>();
    });

    // 和发布端配置完全一致的契约映射
    cfg.Message<Message>(x => {
        x.SetEntityName("shared.message");
        x.SetMessageUrn("urn:message:shared:message");
    });

    cfg.SetMessageSerializer(new JsonMessageSerializer());
});

这种方式通过手动指定消息的实体名称和Urn，让MassTransit把两个不同的类型视为同一个契约，但缺点是需要两边配置完全一致，后期维护容易出错，所以优先推荐方案1。

另外补充一点：_skipped队列是MassTransit的容错机制，当消息无法找到匹配的消费者时，会被自动转发到这个队列，方便你排查未被消费的消息问题。

内容的提问来源于stack exchange，提问作者Алексей Якубин