RabbitMQ集群环境下Prefetch Count设为1未生效的问题求助

我使用RabbitMQ 3.8.14版本和MassTransit 8.0.1版本，部署了3节点RabbitMQ集群。需求是确保每条消息确认（Ack）后，再开始消费下一条消息（队列消息含唯一递增键，需保证同一时刻仅消费一条）。我将Prefetch Count设置为1，但在集群环境压测时，第一条消息未确认就已开始消费第二条消息，与Prefetch Count的预期行为不符。本地非集群环境下则正常，能等待前一条消息确认后再消费下一条。

示例代码 #

public class Caller
{
    ///ctor injections
    private static int number= 0;
   
    public async Task publish()
    {
        await _publishEndpoint.Publish(new GeneratedChipsEvent { messageNumber = ++number});
    }
}

public class GeneratedChipsEvent
{
    public int messageNumber { get; init; }
}

public class GeneratedChipsEventConsumer : IConsumer<GeneratedChipsEvent>
{
    ///ctor injections

    public async Task Consume(ConsumeContext<GeneratedChipsEvent> context)
    {
        _logger.LogInformation($"starting consuming the messsage:{context.Message.messageNumber}");
        await Task.Delay(System.TimeSpan.FromSeconds(3));
        _logger.LogInformation($"ending consuming the messsage:{context.Message.messageNumber}");
    }
}

本地非集群环境日志（正常） #

[10:06:08 INF] starting consuming the messsage:1
[10:06:11 INF] ending consuming the messsage:1
[10:06:11 INF] starting consuming the messsage:2
[10:06:14 INF] ending consuming the messsage:2
[10:06:14 INF] starting consuming the messsage:3
[10:06:17 INF] ending consuming the messsage:3

集群服务器环境日志（异常） #

[10:16:33 INF] starting consuming the messsage:1
[10:16:33 INF] starting consuming the messsage:2
[10:16:33 INF] starting consuming the messsage:3
[10:16:36 INF] ending consuming the messsage:1
[10:16:36 INF] ending consuming the messsage:2
[10:16:36 INF] ending consuming the messsage:3

问题排查与解决办法 #

1. 检查消费者实例数量：集群环境中若启动了多个消费者进程，每个实例的Prefetch Count=1会各自拉取一条消息，导致同时消费。若要严格单条消费，需确保仅运行一个消费者实例，或开启RabbitMQ的独占消费模式（设置队列的x-queue-mode为exclusive）。
2. 明确设置MassTransit并发限制：除了设置PrefetchCount=1，还需配置消费并发数为1，避免同一实例内并行处理。在接收端点配置中添加UseConcurrencyLimit(1)：

   cfg.ReceiveEndpoint("your-target-queue", e =>
   {
       e.PrefetchCount = 1;
       e.UseConcurrencyLimit(1);
       e.Consumer<GeneratedChipsEventConsumer>(provider);
   });
   

3. 排查队列镜像配置：RabbitMQ 3.8.x的镜像队列可能存在预取计数的逻辑异常，可尝试临时将队列改为非镜像队列测试，或升级至RabbitMQ 3.9+版本修复该问题。
4. 确认消息确认逻辑：确保消费完成后自动触发Ack（MassTransit默认在Consume方法完成后自动确认），若有自定义手动确认逻辑，需保证context.Ack()在任务执行完毕后调用。

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