使用MassTransit读取RabbitMQ队列消息失败的问题求助

咱们一步一步拆解你遇到的两个核心问题：

1. 初始代码无输出、消息未被处理的原因 #

你最初的代码只完成了MassTransit总线的配置，但没有启动总线！MassTransit的Bus.Factory.CreateUsingRabbitMq只是生成了总线的配置实例，必须调用bus.Start()才能让总线真正开始监听指定队列、接收并处理消息。

没启动总线的情况下，程序会直接执行完Main方法就退出，自然不会有任何控制台输出，消息也会原封不动留在队列里——因为根本没有消费者在监听。

2. 添加bus.Start()后消息被移到myQueue_skipped的原因 #

这是典型的消息类型不匹配导致的反序列化失败。

MassTransit默认会根据消息类的命名空间、类名、程序集信息生成唯一的类型标识（比如RabbitMQ消息headers里的__TypeId字段），用来识别消息对应的.NET类型。如果队列里的消息是以下情况：

• 用了不同的ProcessingQueue类（比如命名空间不同、类名相同但属性不一致）
• 手动构造的消息（未正确设置MassTransit的类型headers）
• 其他非MassTransit客户端发送的消息

MassTransit就无法将队列中的消息反序列化为你当前定义的ProcessingQueue对象，为了避免消息持续重试阻塞队列，它会把无法处理的消息转移到{队列名}_skipped这个跳过队列中。

修正后的完整代码 #

下面是解决了上述问题的代码，同时添加了错误排查逻辑：

using MassTransit;
using System;

class Program
{
    static void Main(string[] args)
    {
        // 使用using语句确保总线在程序结束时正确释放资源
        using var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
        {
            cfg.Host("localhost", "/", h =>
            {
                h.Username("guest");
                h.Password("guest");
            });

            cfg.ReceiveEndpoint("myQueue", e =>
            {
                // 添加错误处理，方便排查消息处理失败的原因
                e.HandleFault(async context =>
                {
                    Console.WriteLine($"消息处理失败: {context.Fault.Message}");
                    Console.WriteLine($"失败详情: {context.Fault.StackTrace}");
                });

                // 处理ProcessingQueue类型的消息
                e.Handler<ProcessingQueue>(async context =>
                {
                    Console.WriteLine($"收到消息: Id={context.Message.Id}, Name={context.Message.Name}");
                    // MassTransit默认会自动确认消息，除非这里抛出异常
                });

                // 如果你不确定消息类型，可以先启用原始消息处理来查看内容
                // e.Handler<RawMessageContext>(async context =>
                // {
                //     var rawMessage = await context.GetBodyAsString();
                //     Console.WriteLine("原始消息内容: " + rawMessage);
                //     // 查看消息headers里的类型标识
                //     if (context.Headers.TryGetHeader("__TypeId", out var typeId))
                //         Console.WriteLine("消息类型标识: " + typeId);
                // });
            });
        });

        // 启动总线，开始监听队列
        bus.Start();
        Console.WriteLine("已启动消息监听，按回车键停止...");
        Console.ReadLine();

        // 停止总线，释放资源
        bus.Stop();
    }
}

public class ProcessingQueue
{
    public int Id { get; set; }
    public string Name { get; set; }
}

额外排查与修复建议 #

1. 确保消息类型一致性：

   • 如果你是自己发送的消息，确保发送端和接收端使用同一个ProcessingQueue类（最好放在共享类库中，两端都引用），这样类型标识会完全一致。
   • 查看myQueue_skipped队列中消息的__TypeId header，对比它和你当前ProcessingQueue的类型全名（命名空间+类名）是否一致。

2. 处理非MassTransit格式的消息：
   如果队列里的消息不是用MassTransit发送的，你可以：

   • 使用代码中注释的RawMessageContext处理逻辑，直接读取原始消息内容。
   • 配置自定义的消息序列化/反序列化规则，让MassTransit能识别这些消息。

3. 启用日志：
   可以添加日志框架（比如Serilog、NLog）集成MassTransit，查看更详细的运行日志，尤其是反序列化失败的具体错误信息，这会帮你更快定位问题。

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