【RabbitMQ】Basic Queue 简单队列模型与WorkQueue |
您所在的位置：网站首页 › rabbitmq原理详解 › 【RabbitMQ】Basic Queue 简单队列模型与WorkQueue |
【RabbitMQ】Basic Queue 简单队列模型与WorkQueue
|
目录 SpringAMQP 1 Basic Queue 简单队列模型 1.1.消息发送 1.2.消息接收 1.3.测试 2 WorkQueue 2.1.消息发送 2.2.消息接收 2.3.测试 2.4.能者多劳 2.5.总结 SpringAMQPSpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。 SpringAmqp的官方地址：Spring AMQP
SpringAMQP提供了三个功能： 自动声明队列、交换机及其绑定关系 基于注解的监听器模式，异步接收消息 封装了RabbitTemplate工具，用于发送消息 1 Basic Queue 简单队列模型在父工程mq-demo中引入依赖 org.springframework.boot spring-boot-starter-amqp 1.1.消息发送首先配置MQ地址，在publisher服务的application.yml中添加配置： spring: rabbitmq: host: 192.168.150.101 # 主机名 port: 5672 # 端口 virtual-host: / # 虚拟主机 username: itcast # 用户名 password: 123321 # 密码然后在publisher服务中编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送： package cn.itcast.mq.spring; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; @RunWith(SpringRunner.class) @SpringBootTest public class SpringAmqpTest { @Autowired private RabbitTemplate rabbitTemplate; @Test public void testSimpleQueue() { // 队列名称 String queueName = "simple.queue"; // 消息 String message = "hello, spring amqp!"; // 发送消息 rabbitTemplate.convertAndSend(queueName, message); } } 1.2.消息接收首先配置MQ地址，在consumer服务的application.yml中添加配置： spring: rabbitmq: host: 192.168.150.101 # 主机名 port: 5672 # 端口 virtual-host: / # 虚拟主机 username: itcast # 用户名 password: 123321 # 密码然后在consumer服务的cn.itcast.mq.listener包中新建一个类SpringRabbitListener，代码如 下： package cn.itcast.mq.listener; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class SpringRabbitListener { @RabbitListener(queues = "simple.queue") public void listenSimpleQueueMessage(String msg) throws InterruptedException { System.out.println("spring 消费者接收到消息：【" + msg + "】"); } } 1.3.测试启动consumer服务，然后在publisher服务中运行测试代码，发送MQ消息 2 WorkQueueWork queues，也被称为（Task queues），任务模型。简单来说就是让多个消费者绑定到一个队 列，共同消费队列中的消息。
当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。 此时就可以使用work 模型，多个消费者共同处理消息处理，速度就能大大提高了。 2.1.消息发送这次我们循环发送，模拟大量消息堆积现象。 在publisher服务中的SpringAmqpTest类中添加一个测试方法： /** * workQueue * 向队列中不停发送消息，模拟消息堆积。 */ @Test public void testWorkQueue() throws InterruptedException { // 队列名称 String queueName = "simple.queue"; // 消息 String message = "hello, message_"; for (int i = 0; i < 50; i++) { // 发送消息 rabbitTemplate.convertAndSend(queueName, message + i); Thread.sleep(20); } } 2.2.消息接收要模拟多个消费者绑定同一个队列，我们在consumer服务的SpringRabbitListener中添加2个新的 方法： @RabbitListener(queues = "simple.queue") public void listenWorkQueue1(String msg) throws InterruptedException { System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now()); Thread.sleep(20); } @RabbitListener(queues = "simple.queue") public void listenWorkQueue2(String msg) throws InterruptedException { System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now()); Thread.sleep(200); }注意到这个消费者sleep了1000秒，模拟任务耗时。 2.3.测试启动ConsumerApplication后，在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。 可以看到消费者1很快完成了自己的25条消息。消费者2却在缓慢的处理自己的25条消息。 也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。这样显然是有问题的。 2.4.能者多劳在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文 件，添加配置： spring: rabbitmq: listener: simple: prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息 2.5.总结Work模型的使用： 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理 通过设置prefetch来控制消费者预取的消息数量 |
【本文地址】
今日新闻 |
推荐新闻 |