【RabbitMQ】Basic Queue 简单队列模型与WorkQueue
| 阿里云国内75折 回扣 微信号：monov8 |
| 阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6 |
目录
SpringAMQP
SpringAMQP是基于RabbitMQ封装的一套模板并且还利用SpringBoot对其实现了自动装配使用起来非常方便。
SpringAmqp的官方地址Spring AMQP
SpringAMQP提供了三个功能
-
自动声明队列、交换机及其绑定关系
-
基于注解的监听器模式异步接收消息
-
封装了RabbitTemplate工具用于发送消息
1 Basic Queue 简单队列模型
在父工程mq-demo中引入依赖
<!--AMQP依赖包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>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 WorkQueue
Work 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来控制消费者预取的消息数量