RabbitMQ如何实现消费者ACK前暂停向其他消费者投递消息

这个需求没达到预期，一部分来自对RabbitMQ投递机制的认知偏差，一部分来自现有代码的明显错误，以下是具体说明：

你现有代码的致命问题 #

• Qos配置位置错误：BasicQos是消费者侧的信道配置，作用是控制Broker向该信道消费者投递未ACK消息的上限，你把这个配置写在生产者端完全不生效。
• 消费逻辑完全混乱：你创建了EventingBasicConsumer并绑定了Received事件，但从未调用channel.BasicConsume()将消费者注册到队列，绑定的事件永远不会触发；你在循环中调用的channel.BasicGet()是同步单次拉取方法，既没有接收返回结果处理消息，也没有实现长驻拉取逻辑，属于无效调用。
• 资源释放逻辑错误：你在Main方法中调用完接收方法就直接关闭信道和连接，消费者根本无法长驻处理消息。
• 长耗时模拟参数错误：Thread.Sleep(200000000)的单位是毫秒，对应时长接近56小时，完全不符合你要模拟的10秒处理耗时，10秒对应的参数是10000。

需求实现的能力边界说明 #

首先明确RabbitMQ原生多消费者的投递逻辑：Broker会轮询向所有未达到预取配额上限的消费者投递消息，只要消费者还有可用配额，就会持续投递，不会等待上一个消费者完成ACK再给下一个消费者投。
针对你的需求分两种场景判断：

1. 核心需求是「单条消息处理期间，队列不向其他任何消费者投递消息，同一时间全队列只有1条消息在处理，消费者仅做故障转移」
   这个需求完全在RabbitMQ能力范围内，你之前尝试的Single Active Consumer（SAC）就是官方提供的标准方案，之前配置没生效大概率是没有在声明队列时正确开启SAC参数。SAC的逻辑是同一时间只允许一个消费者作为活跃节点消费消息，其余消费者作为热备节点，只有活跃节点断连时才会切换，天然满足“当前消费者未ACK前不向其他消费者投递”的要求。
   注意：SAC不会在每条消息处理完成后自动轮询切换消费者，如果你需要的是严格的C1处理一条、C2处理一条的轮询效果，RabbitMQ原生没有对应自动切换能力——这种模式本质是完全串行消费，多消费者没有任何并行性能收益，仅能做高可用，和SAC的适用场景完全匹配，刻意做逐条轮询切换没有实际业务价值。
2. 必须实现「每条消息ACK后自动轮询切换到下一个消费者处理」
   这个效果RabbitMQ原生不支持，需要在业务层自行实现分布式锁逻辑：所有消费者处理消息前先抢全局锁，拿到锁才能拉取消息处理，处理完释放锁。但这种方案本质还是串行消费，和单消费者+SAC的性能、可用性表现一致，没有额外收益。

可直接运行的正确实现代码 #

队列声明注意事项 #

所有连接（生产者、所有消费者）声明队列时参数必须完全一致，开启SAC的配置如下：

var queueArgs = new Dictionary<string, object>
{
    // 开启单活跃消费者
    {"x-single-active-consumer", true}
};
// 声明持久化队列
channel.QueueDeclare(queue: "Q1", durable: true, exclusive: false, autoDelete: false, arguments: queueArgs);

生产者端代码 #

public static void Main(string[] args)
{
    var factory = new ConnectionFactory() { UserName = "guest", Password = "guest", HostName = "localhost" };
    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();
    
    // 声明队列，参数和上述一致
    var queueArgs = new Dictionary<string, object> {{"x-single-active-consumer", true}};
    channel.QueueDeclare("Q1", true, false, false, queueArgs);

    var properties = channel.CreateBasicProperties();
    properties.Persistent = true;

    int i = 0;
    Console.WriteLine("按回车发送消息");
    while (Console.ReadLine() != null)
    {
        i++;
        var message = $"{DateTime.Now} - 消息{i}";
        var body = Encoding.UTF8.GetBytes(message);
        channel.BasicPublish(exchange: "", routingKey: "Q1", basicProperties: properties, body: body);
        Console.WriteLine($"已发送: {message}");
    }
}

消费者端代码（C1、C2代码完全一致） #

public static async Task Main(string[] args)
{
    var factory = new ConnectionFactory() { HostName = "localhost", UserName = "guest", Password = "guest" };
    using var connection = factory.CreateConnection();
    using var channel = connection.CreateModel();
    
    // 声明队列，参数必须和生产者完全一致
    var queueArgs = new Dictionary<string, object> {{"x-single-active-consumer", true}};
    channel.QueueDeclare("Q1", true, false, false, queueArgs);

    // 消费者端配置Qos：每个信道最多同时持有1条未ACK消息
    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

    var consumer = new EventingBasicConsumer(channel);
    consumer.Received += (sender, ea) =>
    {
        try
        {
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            Console.WriteLine($"消费者{Environment.ProcessId} 收到消息: {message}，时间: {DateTime.Now}");
            
            // 模拟10秒业务处理耗时
            Thread.Sleep(10000);
            
            // 处理完成发送ACK
            channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
            Console.WriteLine($"消费者{Environment.ProcessId} 已确认消息，时间: {DateTime.Now}\n");
        }
        catch (Exception ex)
        {
            // 处理异常时NACK消息，让消息重回队列
            channel.BasicNack(deliveryTag: ea.DeliveryTag, multiple: false, requeue: true);
        }
    };

    // 注册消费者启动消费，不要混用BasicGet
    channel.BasicConsume(queue: "Q1", autoAck: false, consumer: consumer);
    
    Console.WriteLine("消费者已启动，等待消息...");
    // 阻塞等待，不要提前关闭信道和连接
    await Task.Delay(Timeout.Infinite);
}

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