RabbitMQ单消费者应用多消费线程未生效问题排查

我尝试在单个消费者应用中使用多消费线程，启动3个线程调用ConsumeMessages函数，但最终只有单个线程处理所有消息，日志显示所有消息处理的线程ID均为5。

Main函数代码 #

static void Main(string[] args)
{
    Console.WriteLine("Starting");

    // 定义消费线程数量
    int numberOfConsumers = 3;

    for (int i = 0; i < numberOfConsumers; i++)
    {
        // 为每个消费者启动新线程
       var thread = new Thread(new ThreadStart(ConsumeMessages));
        thread.Start();
    }

    var waitHandle = new ManualResetEvent(false);
    waitHandle.WaitOne();
}

ConsumeMessages函数代码 #

private static void ConsumeMessages()
{
    var factory = new ConnectionFactory { HostName = "localhost" };

    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();
    channel.BasicQos(0, 1, false); // 每个消费者预取1条消息

    Console.WriteLine($"Create channel, Thread Id: {Thread.CurrentThread.ManagedThreadId}");

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

    var consumer = new EventingBasicConsumer(channel);

    Console.WriteLine($"Create consumer, Thread Id: {Thread.CurrentThread.ManagedThreadId}");

    consumer.Received += async (model, ea) =>
    {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        var orderInfo = JsonSerializer.Deserialize<OrderDto>(message);
        Console.WriteLine($"Order is Being Processed, Thread Id:    {Thread.CurrentThread.ManagedThreadId}");
        await OrderCheckOut(orderInfo!);
        Console.WriteLine($"Order Processed");
    };

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

注：ConsumeMessages函数仅调用用于更新操作的OrderCheckOut方法。

问题根源 #

1. 连接/通道被提前释放：ConsumeMessages中使用using var声明的connection和channel会在函数执行完毕后自动释放。由于BasicConsume是非阻塞调用，函数执行到最后一行就会退出，连接和通道被销毁，仅第一个线程的资源可能未被及时回收，导致只有它能持续消费。
2. 自动确认导致消息集中分发：autoAck: true会让RabbitMQ在分发消息后立刻标记为已确认，加上异步回调的线程池调度特性，容易造成消息集中流向第一个建立的消费者。

修复方案 #

1. 防止连接/通道提前释放 #

移除using声明，或在函数末尾添加阻塞逻辑，确保线程在消费期间保持存活，避免连接和通道被销毁。

2. 改用手动消息确认 #

将autoAck设为false，在消息处理完成后手动调用BasicAck，让RabbitMQ能公平地将消息分发给多个消费者。

修改后的ConsumeMessages函数 #

private static void ConsumeMessages()
{
    var factory = new ConnectionFactory { HostName = "localhost" };

    // 移除using，确保连接/通道在消费周期内有效
    var connection = factory.CreateConnection();
    var channel = connection.CreateModel();
    channel.BasicQos(0, 1, false); // 每个消费者预取1条消息

    Console.WriteLine($"Create channel, Thread Id: {Thread.CurrentThread.ManagedThreadId}");

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

    var consumer = new EventingBasicConsumer(channel);

    Console.WriteLine($"Create consumer, Thread Id: {Thread.CurrentThread.ManagedThreadId}");

    consumer.Received += async (model, ea) =>
    {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        var orderInfo = JsonSerializer.Deserialize<OrderDto>(message);
        Console.WriteLine($"Order is Being Processed, Thread Id: {Thread.CurrentThread.ManagedThreadId}");
        
        await OrderCheckOut(orderInfo!);
        
        // 处理完成后手动确认消息
        channel.BasicAck(ea.DeliveryTag, false);
        Console.WriteLine($"Order Processed");
    };

    // 设置autoAck为false，启用手动确认
    channel.BasicConsume(queue: "MS", autoAck: false, consumer: consumer);

    // 阻塞线程，防止函数退出导致资源释放
    Console.ReadLine();
}

3. 优化主线程等待逻辑 #

原Main函数中的ManualResetEvent可以保留，确保应用不会提前退出；消费线程通过Console.ReadLine()保持存活，避免连接/通道被销毁。

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