MessageQueue --- RabbitMQ WorkQueue

什么是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秒。这样显然是有问题的。

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 = connection.createChannel();
channel.basicQos(1); // 设置预取计数为 1
channel.basicConsume("queue_name", false, consumer); 
// 关闭自动确认（autoAck=false）

• 消费者在处理完当前消息并手动确认（ack）后，才会接收下一条消息。

特点：

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

Example

• 在spring中有一个简单的配置，可以解决这个问题。我们修改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条消息