.NET 6中RabbitMQ.Client v6.4.0消费队列无数据返回问题排查

核心问题 #

1. 事件触发顺序错误：你在调用channel.BasicConsume()之前就检查message是否为空，而Received事件只有在调用BasicConsume并成功订阅队列后才会触发，此时主线程已经走完判断逻辑，message自然是null。
2. 连接/通道生命周期过短：connection和channel没有用using包裹，也没有保持长期存活，当RabbitMQConsumer方法执行完毕后，这两个对象会被GC回收，导致消费者连接直接断开，根本没机会接收消息。
3. 异步逻辑处理不当：Received事件是异步回调，主线程和回调线程的变量赋值不同步，主线程无法等待回调完成就执行判断。

修正后的代码实现 #

方案1：在应用启动时初始化长期消费者（推荐）

Web API中不应该每次请求创建消费者，应该在应用启动时初始化一个长期运行的消费者，这样才能持续监听队列：

// 在Program.cs中添加消费者初始化逻辑
var builder = WebApplication.CreateBuilder(args);

// 注册RabbitMQ连接和通道（单例）
builder.Services.AddSingleton<IConnection>(sp =>
{
    var factory = new ConnectionFactory { HostName = "localhost" };
    return factory.CreateConnection();
});

builder.Services.AddSingleton<IModel>(sp =>
{
    var connection = sp.GetRequiredService<IConnection>();
    var channel = connection.CreateModel();
    channel.QueueDeclare(
        queue: "consumption",
        durable: false,
        exclusive: false,
        autoDelete: false,
        arguments: null);
    return channel;
});

var app = builder.Build();

// 启动消费者
var channel = app.Services.GetRequiredService<IModel>();
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    var body = ea.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);
    // 这里处理消息，比如调用UpdateTimestamps
    UpdateTimestamps(message);
};
channel.BasicConsume(queue: "consumption", autoAck: true, consumer: consumer);

// 其他Web API配置
app.MapPost("/api/consumer", () => Results.Ok("消费者已启动"));

app.Run();

// 你的UpdateTimestamps方法
void UpdateTimestamps(string message)
{
    // 实现消息处理逻辑
    Console.WriteLine($"收到消息：{message}");
}

方案2：修正原请求方法的逻辑（仅用于测试，不推荐生产环境）

如果只是临时测试，修改原RabbitMQConsumer方法，确保等待消息接收（注意：这种方式不适合生产，因为Web API请求会挂起）：

[HttpPost]
public async Task RabbitMQConsumer(Timestamp timestampRequest)
{
    using var factory = new ConnectionFactory { HostName = "localhost" };
    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();

    channel.QueueDeclare(
        queue: "consumption",
        durable: false,
        exclusive: false,
        autoDelete: false,
        arguments: null);

    var tcs = new TaskCompletionSource<bool>();
    var consumer = new EventingBasicConsumer(channel);
    
    consumer.Received += (model, ea) =>
    {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        UpdateTimestamps(message);
        tcs.SetResult(true); // 收到消息后完成任务
    };

    channel.BasicConsume(queue: "consumption", autoAck: true, consumer: consumer);

    // 等待收到一条消息（可设置超时）
    await tcs.Task.WaitAsync(TimeSpan.FromSeconds(10));
}

额外注意事项 #

• 避免每次请求创建消费者：生产环境中，消费者应该是长期运行的单例，否则会导致重复订阅、资源浪费，甚至消息丢失。
• 消息确认机制：如果需要保证消息不丢失，不要用autoAck: true，应该在处理完消息后手动调用channel.BasicAck(ea.DeliveryTag, false);。
• 异常处理：在Received事件回调中添加异常捕获，避免单个消息处理失败导致消费者崩溃。

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