Work queues,也被称为(Task queues),任务模型。简单来说就是让多个消费者绑定到一个队列,共同消费队列中的消息
image-20210717164238910.png
当消息处理比较耗时的时候,可能生产消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多,无法及时处理。
此时就可以使用work 模型,多个消费者共同处理消息处理,速度就能大大提高了。

1.引入依赖

Basic Queue

2.消息发送

这次我们循环发送,模拟大量消息堆积现象。
在publisher服务中的SpringAmqpTest类中添加一个测试方法:

  1. /**
  2. * workQueue
  3. * 向队列中不停发送消息,模拟消息堆积。
  4. */
  5. @Test
  6. public void testWorkQueue() throws InterruptedException {
  7. // 队列名称
  8. String queueName = "simple.queue";
  9. // 消息
  10. String message = "hello, message_";
  11. for (int i = 0; i < 50; i++) {
  12. // 发送消息
  13. rabbitTemplate.convertAndSend(queueName, message + i);
  14. Thread.sleep(20);
  15. }
  16. }

3.消息接受

要模拟多个消费者绑定同一个队列,我们在consumer服务的SpringRabbitListener中添加2个新的方法:

  1. package cn.itcast.mq.listener;
  2. @Component
  3. public class SpringRabbitListener {
  4. @RabbitListener(queues = "simple.queue")
  5. public void listenWorkQueue1(String msg) throws InterruptedException {
  6. System.out.println("消费者1接收到消息:【" + msg + "】" + LocalTime.now());
  7. Thread.sleep(20);
  8. }
  9. @RabbitListener(queues = "simple.queue")
  10. public void listenWorkQueue2(String msg) throws InterruptedException {
  11. System.err.println("消费者2........接收到消息:【" + msg + "】" + LocalTime.now());
  12. Thread.sleep(200);
  13. }
  14. }

4.测试

启动ConsumerApplication后,在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。
可以看到消费者1很快完成了自己的25条消息。消费者2却在缓慢的处理自己的25条消息。
也就是说消息是平均分配给每个消费者,并没有考虑到消费者的处理能力
image.png

消息预取机制

就像上面那样,提前先分配好,两个消费者要处理哪些消息。把这些消息先给分配的消费者,等他们慢慢处理。所以消费者2就消息堆积了。
在spring中有一个简单的配置,可以解决这个问题。我们修改consumer服务的application.yml文件,添加配置:

  1. spring:
  2. rabbitmq:
  3. listener:
  4. simple:
  5. prefetch: 1 # 每次只能获取一条消息,处理完成才能获取下一个消息