MessageQueue --- RabbitMQ WorkQueue and Prefetch

什么是WorkQueue

• Work queues，任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。
• 当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用workqueu模型，多个消费者共同处理消息处理，消息处理的速度就能大大提高了。

分发机制 — RoundRobin

工作机制：

• 默认模式：当多个消费者订阅同一个队列时，RabbitMQ 会依次将消息分发给每个消费者，按顺序循环分配。
• 示例：
  队列中有消息 M1, M2, M3, M4，消费者 C1 和 C2 同时订阅。
  分发顺序为：M1 → C1，M2 → C2，M3 → C1，M4 → C2。

特点：

• 简单高效：无需额外配置，适合消费者处理速度相近的场景。

潜在问题：

• 若消费者处理速度差异较大，可能导致某些消费者空闲，而其他消费者积压消息。
• 例如：C1 处理速度慢，C2 处理速度快，但 C1 仍会分配到一半的消息，造成负载不均衡。

Example

//消息发送
//循环发送，模拟大量消息堆积现象。
@Test
public void testWorkQueue() throws InterruptedException {// 队列名称String queueName = "simple.queue";// 消息String message = "hello, message_";for (int i = 0; i < 50; i++) {// 发送消息，每20毫秒发送一次，相当于每秒发送50条消息rabbitTemplate.convertAndSend(queueName, message + i);Thread.sleep(20);}
}

//消息接收
//模拟多个消费者绑定同一个队列，我们添加2个方法，
//并且设置不同睡眠时间模拟不同性能读取
@RabbitListener(queues = "work.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());Thread.sleep(20);
}
@RabbitListener(queues = "work.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());Thread.sleep(200);
}

• 消费者1很快完成了自己的25条消息
• 消费者2却在缓慢的处理自己的25条消息。
• 也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。导致1个消费者空闲，另一个消费者忙的不可开交。没有充分利用每一个消费者的能力，最终消息处理的耗时远远超过了1秒。这样显然是有问题的。

下面我们介绍prefetch机制，可以做到fair dispatch

分发机制 — Prefetch

工作机制：

• 配置预取计数（Prefetch Count）：通过设置 basicQos 参数，限制每个消费者未确认（unacknowledged）的消息数量。

• 进入prefetch的消息仍会被保留在队列中，但是同时也会发给消费者等待处理
  在 RabbitMQ 的原始队列（Queue）中，会被标记为 “Unacked”（未确认）状态。
  这些消息不会被其他消费者获取（即使设置了 prefetch 的消费者崩溃）。
  只有消费者显式发送 ack 或 nack 后，消息才会从队列中移除（或重新排队）。

消息状态变化流程

• 消息推送给消费者：
  RabbitMQ 将消息标记为 “Unacked”，但仍在队列中（占用内存或磁盘，取决于队列持久化配置）。
  此时消息对其他消费者不可见。
• 消费者处理消息：
  若成功处理并发送 ack → 消息从队列中物理删除。
  若发送 nack(requeue=true) → 消息重新变为 “Ready” 状态，可被其他消费者获取。
  若发送 nack(requeue=false) 或者超时→ 消息被放入死信队列，如果没有配置死信队列则被丢弃

示例：

Channel channel = ...;
Consumer consumer = ...;
channel.basicQos(10); // Per consumer limit
channel.basicConsume("my-queue", false, consumer);

Channel channel = ...;
Consumer consumer = ...;
channel.basicQos(0); // No limit for this consumer，allowing any number of unacknowledged messages.
channel.basicConsume("my-queue", false, consumer);

Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10); // Per consumer limit
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);

Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10, false); // Per consumer limit
channel.basicQos(15, true);  // Per channel limit
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);
//这两个消费者之间总共最多只能有 15 条未确认消息，且每个消费者最多处理 10 条消息。
//由于需要在 Channel 和队列之间协调全局限制，该模式的性能会低于前述示例（存在额外开销）

特点：

• 负载均衡：处理速度快的消费者会获取更多消息，避免空闲
• 可以一次性发送多个消息给消费者处理，减少网络开销
• 可靠性：需配合手动确认（ack）机制，确保消息处理成功后才从队列移除。
• 适用场景：消费者处理速度差异较大时（如耗时任务），能显著提升整体吞吐量。
• Automatic acknowledgement mode or manual acknowledgement mode with unlimited prefetch should be used with care. 通常设为 100~300，平衡吞吐与内存占用。

Note:

• AMQP 0-9-1 协议是channel level prefetch，通过 basic.qos 方法限制channel上的未确认消息数
• channel level有很大缺陷，由于单个channel可能从多个queue消费消息，channel与queue之间需要为每条消息进行协调，以确保不超出限制。这种机制在单机环境下效率较低，而在集群消费场景中性能会显著下降，大多数使用场景也需要consumer level prefetch
• 所以RabbitMQ支持consumer level prefetch (也就是以上的例子)
  

Spring example use prefetch — Fair Dispatch

• 在spring中有一个prefetch的配置，我们修改consumer服务的application.yml文件，添加配置：

spring:rabbitmq:listener:simple:acknowledge-mode: manual # 设置确认方式为手动确认prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

• 可以发现，由于消费者1处理速度较快，所以处理了更多的消息；消费者2处理速度较慢，只处理了6条消息。而最终总的执行耗时也在1秒左右，大大提升

• 还可根据实际情况自定义prefetch count，达到限流的目的

spring:rabbitmq:listener:simple:acknowledge-mode: manual # 设置确认方式为手动确认prefetch: 5 # 限制消费者只能接收5条消息