微服务框架 SpringCloud微服务架构 16 SpringAMQP 16.4 WorkQueue模型

微服务框架

【SpringCloud+RabbitMQ+Docker+Redis+搜索+分布式，系统详解springcloud微服务技术栈课程|黑马程序员Java微服务】

SpringCloud微服务架构

16 SpringAMQP

16.4 WorkQueue模型

16.4.1 WorkQueue 工作队列

Work queue，工作队列，可以提高消息处理速度，避免队列消息堆积

WorkQueue 工作队列结构如下

可以看到和第一个模型不一样的地方是消息队列后面挂了两个消费者【加快处理】

16.4.2 案例

模拟WorkQueue，实现一个队列绑定多个消费者

基本思路如下：

1. 在publisher服务中定义测试方法，每秒产生50条消息，发送到simple.queue

@Test
public void testSendMessage2WorkQueue() throws InterruptedException {
    String queueName = "simple.queue";
    String message = "Hello,DingJiaxiong__";
    for (int i = 1; i <= 50; i++) {
        rabbitTemplate.convertAndSend(queueName,message + i);
        Thread.sleep(20);
    }
    
}
1
2
3
4
5
6
7
8
9
10

2. 在consumer服务中定义两个消息监听者，都监听simple.queue队列
3. 消费者1每秒处理50条消息，消费者2每秒处理10条消息

@RabbitListener(queues = "simple.queue")
public void listenerWorkQueue1(String msg) throws InterruptedException {
    
    System.out.println("消费者1接收到simple.queue的消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(20);
}

@RabbitListener(queues = "simple.queue")
public void listenerWorkQueue2(String msg) throws InterruptedException {
    System.err.println("消费者2接收到simple.queue的消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(200);
}
1
2
3
4
5
6
7
8
9
10
11
12

直接重启这个服务

直接运行生产者测试，生产消息

OK， 50条消息是生产完成了，看看那边的日志

OK，19开始，24结束， 五秒钟内完成了所有消息的消费…【这和我们预想的不太一样】

仔细看看日志

消费者1 用了2秒左右，但是消费者2 就长了【19 - 24】

在我们写的两个消费者中，

一个20ms，一个200ms，很明显消费者1 是要处理得更快的，它更快，按理说应该多消费两条

，但是实际情况不是这样， 实际情况是两个消费者各自消费了25 条

【这种分配方式并没有考虑不同消费者的消费能力】

【原因】RabbitMQ 内部的消息预取机制，当有大量消息到达队列，它会先把消息进行统一，两个消费者的通道会提前把消息拿过来【这个就是消息预取】，不管能不能处理，先拿过来再说，于是这俩消费者就每个平分 了25 条

【所以可以进行控制吗？不能处理不要拿这么多】 【当然】

【消费预期限制】

修改application.yml文件，设置preFetch这个值，可以控制预取消息的上限：

logging:
  pattern:
    dateformat: MM-dd HH:mm:ss:SSS

spring:
  rabbitmq:
    host: 1.13.13.44 # rabbitMQ 的服务IP地址
    port: 5672 # 端口
    username: itcast
    password: 123321
    virtual-host: /

    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

OK， 到现在重启消费者

再来一次，再次生产50 条消息到队列中

OK，生产完成，查看消费者的日志

虽然还是没有在1秒内完成，但是这次明显可以看到消费者1 消费了更多的消息

但是笔者这里和老师的不太一样，按理说，消费者1应该比消费者2 快10倍，但是运行结果并没有

【算了不管了，感觉和电脑性能也有关系，笔者的控制台根本刷不到老师那么快】

16.4.3 总结

Work模型的使用：

• 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
• 通过设置prefetch来控制消费者预取的消息数量