RabbitMQ:WorkQueues模型

简介: RabbitMQ:WorkQueues模型

♥️作者:小宋1021

🤵‍♂️个人主页:小宋1021主页

♥️坚持分析平时学习到的项目以及学习到的软件开发知识,和大家一起努力呀!!!

🎈🎈加油! 加油! 加油! 加油

🎈欢迎评论 💬点赞👍🏻 收藏 📂加关注+


Work queues,任务模型。简单来说就是多个消费者绑定到一个队列,共同消费队列中的消息

image.gif 编辑

当消息处理比较耗时的时候,可能生产消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多,无法及时处理。

此时就可以使用work 模型,多个消费者共同处理消息处理,消息处理的速度就能大大提高了。

接下来,我们就来模拟这样的场景。

基本思路如下:

1.在RabbitMQ的控制台创建一个队列,名为work.queue

2.在publisher服务中定义测试方法,发送50条消息到work.queue

3.在consumer服务中定义两个消息监听者,都监听work.queue队列

首先,我们在控制台创建一个新的队列,命名为work.queue

image.gif 编辑

RabbitMQ对应的架构如图:

image.gif 编辑

其中包含几个概念:

  • publisher:生产者,也就是发送消息的一方
  • consumer:消费者,也就是消费消息的一方
  • queue:队列,存储消息。生产者投递的消息会暂存在消息队列中,等待消费者处理
  • exchange:交换机,负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。
  • virtual host:虚拟主机,起到数据隔离的作用。每个虚拟主机相互独立,有各自的exchange、queue

上述这些东西都可以在RabbitMQ的管理控制台来管理,下一节我们就一起来学习控制台的使用。

消息发送

这次我们循环发送,模拟大量消息堆积现象。

在publisher服务中的SpringAmqpTest类中添加一个测试方法:

/**
     * workQueue
     * 向队列中不停发送消息,模拟消息堆积。
     */
@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);
    }
}

image.gif

消息接收

要模拟多个消费者绑定同一个队列,我们在consumer服务的SpringRabbitListener中添加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);
}

image.gif

注意到这两消费者,都设置了Thead.sleep,模拟任务耗时:

  • 消费者1 sleep了20毫秒,相当于每秒钟处理50个消息
  • 消费者2 sleep了200毫秒,相当于每秒处理5个消息

测试

启动ConsumerApplication后,在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。

最终结果如下:

消费者1接收到消息:【hello, message_0】21:06:00.869555300

消费者2........接收到消息:【hello, message_1】21:06:00.884518

消费者1接收到消息:【hello, message_2】21:06:00.907454400

消费者1接收到消息:【hello, message_4】21:06:00.953332100

消费者1接收到消息:【hello, message_6】21:06:00.997867300

消费者1接收到消息:【hello, message_8】21:06:01.042178700

消费者2........接收到消息:【hello, message_3】21:06:01.086478800

消费者1接收到消息:【hello, message_10】21:06:01.087476600

消费者1接收到消息:【hello, message_12】21:06:01.132578300

消费者1接收到消息:【hello, message_14】21:06:01.175851200

消费者1接收到消息:【hello, message_16】21:06:01.218533400

消费者1接收到消息:【hello, message_18】21:06:01.261322900

消费者2........接收到消息:【hello, message_5】21:06:01.287003700

消费者1接收到消息:【hello, message_20】21:06:01.304412400

消费者1接收到消息:【hello, message_22】21:06:01.349950100

消费者1接收到消息:【hello, message_24】21:06:01.394533900

消费者1接收到消息:【hello, message_26】21:06:01.439876500

消费者1接收到消息:【hello, message_28】21:06:01.482937800

消费者2........接收到消息:【hello, message_7】21:06:01.488977100

消费者1接收到消息:【hello, message_30】21:06:01.526409300

消费者1接收到消息:【hello, message_32】21:06:01.572148

消费者1接收到消息:【hello, message_34】21:06:01.618264800

消费者1接收到消息:【hello, message_36】21:06:01.660780600

消费者2........接收到消息:【hello, message_9】21:06:01.689189300

消费者1接收到消息:【hello, message_38】21:06:01.705261

消费者1接收到消息:【hello, message_40】21:06:01.746927300

消费者1接收到消息:【hello, message_42】21:06:01.789835

消费者1接收到消息:【hello, message_44】21:06:01.834393100

消费者1接收到消息:【hello, message_46】21:06:01.875312100

消费者2........接收到消息:【hello, message_11】21:06:01.889969500

消费者1接收到消息:【hello, message_48】21:06:01.920702500

消费者2........接收到消息:【hello, message_13】21:06:02.090725900

消费者2........接收到消息:【hello, message_15】21:06:02.293060600

消费者2........接收到消息:【hello, message_17】21:06:02.493748

消费者2........接收到消息:【hello, message_19】21:06:02.696635100

消费者2........接收到消息:【hello, message_21】21:06:02.896809700

消费者2........接收到消息:【hello, message_23】21:06:03.099533400

消费者2........接收到消息:【hello, message_25】21:06:03.301446400

消费者2........接收到消息:【hello, message_27】21:06:03.504999100

消费者2........接收到消息:【hello, message_29】21:06:03.705702500

消费者2........接收到消息:【hello, message_31】21:06:03.906601200

消费者2........接收到消息:【hello, message_33】21:06:04.108118500

消费者2........接收到消息:【hello, message_35】21:06:04.308945400

消费者2........接收到消息:【hello, message_37】21:06:04.511547700

消费者2........接收到消息:【hello, message_39】21:06:04.714038400

消费者2........接收到消息:【hello, message_41】21:06:04.916192700

消费者2........接收到消息:【hello, message_43】21:06:05.116286400

消费者2........接收到消息:【hello, message_45】21:06:05.318055100

消费者2........接收到消息:【hello, message_47】21:06:05.520656400

消费者2........接收到消息:【hello, message_49】21:06:05.723106700

可以看到消费者1和消费者2竟然每人消费了25条消息:

  • 消费者1很快完成了自己的25条消息
  • 消费者2却在缓慢的处理自己的25条消息。

也就是说消息是平均分配给每个消费者,并没有考虑到消费者的处理能力。导致1个消费者空闲,另一个消费者忙的不可开交。没有充分利用每一个消费者的能力,最终消息处理的耗时远远超过了1秒。这样显然是有问题的。

能者多劳

在spring中有一个简单的配置,可以解决这个问题。我们修改consumer服务的application.yml文件,添加配置:

spring:

 rabbitmq:

   listener:

     simple:

       prefetch: 1 # 每次只能获取一条消息,处理完成才能获取下一个消息

再次测试,发现结果如下:

消费者1接收到消息:【hello, message_0】21:12:51.659664200

消费者2........接收到消息:【hello, message_1】21:12:51.680610

消费者1接收到消息:【hello, message_2】21:12:51.703625

消费者1接收到消息:【hello, message_3】21:12:51.724330100

消费者1接收到消息:【hello, message_4】21:12:51.746651100

消费者1接收到消息:【hello, message_5】21:12:51.768401400

消费者1接收到消息:【hello, message_6】21:12:51.790511400

消费者1接收到消息:【hello, message_7】21:12:51.812559800

消费者1接收到消息:【hello, message_8】21:12:51.834500600

消费者1接收到消息:【hello, message_9】21:12:51.857438800

消费者1接收到消息:【hello, message_10】21:12:51.880379600

消费者2........接收到消息:【hello, message_11】21:12:51.899327100

消费者1接收到消息:【hello, message_12】21:12:51.922828400

消费者1接收到消息:【hello, message_13】21:12:51.945617400

消费者1接收到消息:【hello, message_14】21:12:51.968942500

消费者1接收到消息:【hello, message_15】21:12:51.992215400

消费者1接收到消息:【hello, message_16】21:12:52.013325600

消费者1接收到消息:【hello, message_17】21:12:52.035687100

消费者1接收到消息:【hello, message_18】21:12:52.058188

消费者1接收到消息:【hello, message_19】21:12:52.081208400

消费者2........接收到消息:【hello, message_20】21:12:52.103406200

消费者1接收到消息:【hello, message_21】21:12:52.123827300

消费者1接收到消息:【hello, message_22】21:12:52.146165100

消费者1接收到消息:【hello, message_23】21:12:52.168828300

消费者1接收到消息:【hello, message_24】21:12:52.191769500

消费者1接收到消息:【hello, message_25】21:12:52.214839100

消费者1接收到消息:【hello, message_26】21:12:52.238998700

消费者1接收到消息:【hello, message_27】21:12:52.259772600

消费者1接收到消息:【hello, message_28】21:12:52.284131800

消费者2........接收到消息:【hello, message_29】21:12:52.306190600

消费者1接收到消息:【hello, message_30】21:12:52.325315800

消费者1接收到消息:【hello, message_31】21:12:52.347012500

消费者1接收到消息:【hello, message_32】21:12:52.368508600

消费者1接收到消息:【hello, message_33】21:12:52.391785100

消费者1接收到消息:【hello, message_34】21:12:52.416383800

消费者1接收到消息:【hello, message_35】21:12:52.439019

消费者1接收到消息:【hello, message_36】21:12:52.461733900

消费者1接收到消息:【hello, message_37】21:12:52.485990

消费者1接收到消息:【hello, message_38】21:12:52.509219900

消费者2........接收到消息:【hello, message_39】21:12:52.523683400

消费者1接收到消息:【hello, message_40】21:12:52.547412100

消费者1接收到消息:【hello, message_41】21:12:52.571191800

消费者1接收到消息:【hello, message_42】21:12:52.593024600

消费者1接收到消息:【hello, message_43】21:12:52.616731800

消费者1接收到消息:【hello, message_44】21:12:52.640317

消费者1接收到消息:【hello, message_45】21:12:52.663111100

消费者1接收到消息:【hello, message_46】21:12:52.686727

消费者1接收到消息:【hello, message_47】21:12:52.709266500

消费者2........接收到消息:【hello, message_48】21:12:52.725884900

消费者1接收到消息:【hello, message_49】21:12:52.746299900

可以看到,在订阅模型中,多了一个exchange角色,而且过程略有变化:

  • Publisher:生产者,不再发送消息到队列中,而是发给交换机
  • Exchange:交换机,一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。
  • Queue:消息队列也与以前一样,接收消息、缓存消息。不过队列一定要与交换机绑定。
  • Consumer:消费者,与以前一样,订阅队列,没有变化

Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!

交换机的类型有四种:

  • Fanout:广播,将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
  • Direct:订阅,基于RoutingKey(路由key)发送给订阅了消息的队列
  • Topic:通配符订阅,与Direct类似,只不过RoutingKey可以使用通配符
  • Headers:头匹配,基于MQ的消息头匹配,用的较少。
相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
5月前
|
消息中间件
RabbitMQ消息模型之Work Queues
RabbitMQ消息模型之Work Queues
35 1
RabbitMQ消息模型之Work Queues
|
5月前
|
消息中间件
RabbitMQ消息模型之Routing-Topic
RabbitMQ消息模型之Routing-Topic
38 0
|
5月前
|
消息中间件 缓存
RabbitMQ消息模型之Sample
RabbitMQ消息模型之Sample
32 0
|
5月前
|
消息中间件
RabbitMQ消息模型之Routing-Direct
RabbitMQ消息模型之Routing-Direct
89 1
|
5月前
|
消息中间件
RabbitMQ消息模型之发布订阅Publish-Subscribe
RabbitMQ消息模型之发布订阅Publish-Subscribe
73 0
RabbitMQ消息模型之发布订阅Publish-Subscribe
|
5月前
|
消息中间件 JSON 缓存
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(SpringBoot整合)
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(SpringBoot整合)
138 0
|
2月前
|
消息中间件 测试技术 Kafka
Apache RocketMQ 批处理模型演进之路
RocketMQ 早期批处理模型存在一定的约束条件,为进一步提升性能,RocketMQ 进行了索引构建流水线改造,同时 BatchCQ 模型和 AutoBatch 模型也优化了批处理流程,提供了更简便的使用体验,快点击本文查看详情及配置展示~
19733 75
|
5月前
|
消息中间件 存储 Apache
RocketMQ实战教程之常见概念和模型
Apache RocketMQ 实战教程介绍了其核心概念和模型。消息是基本的数据传输单元,主题是消息的分类容器,支持字节、数字和短划线命名,最长64个字符。消息类型包括普通、顺序、事务和定时/延时消息。消息队列是实际存储和传输消息的容器,是主题的分区。消费者分组是一组行为一致的消费者的逻辑集合,也有命名限制。此外,文档还提到了一些使用约束和建议,如主题和消费者组名的命名规则,消息大小限制,请求超时时间等。RocketMQ 提供了多种消息模型,包括发布/订阅模型,有助于理解和优化消息处理。
|
5月前
|
消息中间件 JSON 缓存
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(基于SpringBoot)
RabbitMQ快速学习之WorkQueues模型、三种交换机、消息转换器(基于SpringBoot)
170 0
|
3月前
|
消息中间件 C语言 RocketMQ
消息队列 MQ操作报错合集之出现"Connection reset by peer"的错误,该如何处理
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。