消息队列MQ——Spring Boot整合RabbitMQ

Posted 2021-10-18 冉木

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了消息队列MQ——Spring Boot整合RabbitMQ相关的知识，希望对你有一定的参考价值。

系列文章目录

消息队列MQ(一)——RabbitMQ的介绍、安装以及管理页面的使用
 消息队列MQ(二)——Spring Boot整合RabbitMQ

前言

MQ全称为Message Queue，消息队列是应用程序和应用程序之间的通信方法。RabbitMQ是一个Erlang开发的AMQP（Advanced Message Queuing Protocol ）的开源实现。

在项目中，可将一些无需即时返回且耗时的操作提取出来，进行异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。

提示：以下是本篇文章正文内容，下面案例可供参考

一、简介

在spring boot项目中，只需要引入start-amqp起步依赖，即可整合RabbitMQ成功；我们基于SpringBoot封装的RabbitTemplate模板对象，可以非常方便的发送消息，接收消息(使用注解)。

amqp的官方GitHub地址：https://github.com/spring-projects/spring-amqp

二、搭建工程

创建一个SpringBoot 父工程，勾选初始依赖；在下面新建两个子工程——Producer和Consumer

1.创建P系统(生产者)

同理创建SpringBoot的生产者工程：rabbitmq-producer

pom.xml文件如下：

<parent>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-parent</artifactId>
	<version>2.1.9.RELEASE</version>
	<relativePath/> <!-- lookup parent from repository -->
</parent>
<dependencies>
	<!--amqp协议的起步依赖坐标-->
	<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
	</dependency>
	
	<!--rabbit测试依赖坐标-->
	<dependency>
	<groupId>org.springframework.amqp</groupId>
	<artifactId>spring-rabbit-test</artifactId>
	<scope>test</scope>
	</dependency>
	
	<!--SpringBoot测试依赖坐标-->
	<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-test</artifactId>
	<scope>test</scope>
	</dependency>
</dependencies>

1.1 yml配置RabbitMQ属性

# RabbitMQ 服务host地址
spring.rabbitmq.host=121.138.152.111

# 端口
spring.rabbitmq.port=5672

# 虚拟主机地址
spring.rabbitmq.virtual-host=/demo

# rabbit服务的用户名
spring.rabbitmq.username=demo

# rabbit服务的密码
spring.rabbitmq.password=demo

2.创建C系统(消费者)

同理创建SpringBoot的消费者工程：rabbitmq-consumer

pom.xml文件如下：

<parent>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-parent</artifactId>
	<version>2.1.9.RELEASE</version>
	<relativePath/> <!-- lookup parent from repository -->
</parent>
<dependencies>
	<!--amqp的起步依赖坐标-->
	<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
	</dependency>
</dependencies>

2.1 yml配置RabbitMQ属性

# RabbitMQ 服务host地址
spring.rabbitmq.host=121.138.152.111

# 端口
spring.rabbitmq.port=5672

# 虚拟主机地址
spring.rabbitmq.virtual-host=/demo

# rabbit服务的用户名
spring.rabbitmq.username=demo

# rabbit服务的密码
spring.rabbitmq.password=demo

三、实操RabbitMQ五种工作模式

五种工作模式是一种迭代关系，
①比如简单模式是一对一；
②那我有两个甚至多个消费者怎么办，这就有了工作队列模式一对多；
③现在我觉得消费者太多，队列太少不够用，于是就引进多个队列，对应不同的消费者，这就有了发布订阅模式；
④现在我只想给其中一个消费者的消息队列发消息，其他的不发怎么办，于是就是有Routing路由模式；
⑤但是问题又来了，我现在想要某一种的消息类型都发到某个队列中去，路由模式一一对应太严苛了，怎么办？就有了通配符(主题)模式。

3.1 HelloWorld简单模式

简介：在上面的模型中，P称为生产者（主动发送消息的程序），C称为消费者（消息的接受者，会一直等消息的到来），两者关系一攻一受(狗头)。quene：红色部分，消息队列。生产者往里投递消息，消费者从中取消息。这就是简单消息队列模式。

3.1.1 RabbitMQ管理界面操作

创建simple_queue消息队列用于演示Hello World简单模式

点击simple_queue进入这个消息队列管理页面

点击 Get_Message可以查看队列中获取的消息

3.1.2 代码编写

rabbitmq-producer项目测试代码如下:

@RunWith(SpringRunner.class)
@SpringBootTest
public class Demo01TestSimpleQueue {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@Test
	public void contextLoads() {
		//向消息队列发送一条简单消息
		/**
		* 参数1：消息队列名称
		* 参数2：消息内容
		*/
		rabbitTemplate.convertAndSend("simple_queue","hello 小兔子！");
	}
}

rabbitmq-consumer项目创建监听器接收消息:

/**
* 消费者，接收消息队列消息监听器
* 必须将当前监听器对象注入Spring的容器中
*/
@Component
@RabbitListener(queues = "simple_queue")
public class SimpleListener {
	@RabbitHandler
	public void simpleHandler(String msg){
		System.out.println("=====接收消息====>"+msg);
	}
}

启动SpringbootRabbitmqConsumerApplication, 就可以接收到RabbitMQ服务器发送来的消息

3.2 Work queues工作队列模式

简介：Work Queue与简单队列模式相比，多了一个或者一些消费端，多个消费端共同消费同一个队列中的消息。本来是一对一，现在一对多，所以这种模式下，消费者们之间对消息队列中的消息是竞争的关系。

3.2.1 RabbitMQ管理界面操作

创建 work_queue 队列用于演示work工作队列模式

3.2.2 代码编写

生产者代码：

@RunWith(SpringRunner.class)
@SpringBootTest
public class Demo02TestWorkQueue {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@Test
	public void contextLoads() {
		//向工作队列发送1千条消息
		for (int i = 0; i < 1000; i++) {
			/**
			* 参数1：消息队列名称
			* 参数2：消息内容
			*/
			rabbitTemplate.convertAndSend("work_queue","hello 我是小兔子
			【"+i+"】！");
		}
	}
}

消费者1代码：

/**
* 工作队列：消费者接收监听器1，接收来自消息队列中的消息
*/
@Component
@RabbitListener(queues = "work_queue")
public class WorkListener1 {
	@RabbitHandler
	public void workHandler(String msg){
		System.out.println("=====工作队列接收消息端1====>"+msg);
	}
}

消费者2代码：

/**
 * 工作队列：消费者接收监听器2，接收来自消息队列中的消息
*/
@Component
@RabbitListener(queues = "work_queue")
public class WorkListener2 {
	@RabbitHandler
	public void workHandler(String msg){
		System.out.println("=====工作队列消息接收端2====>"+msg);
	}
}

3.3 Publish/Subscribe发布与订阅模式

简介：订阅者模型中，多了一个exchange，即交换机。
1、每个消费者监听自己的队列。
2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息
【广播消息：一次性将消息发送给所有消费者，每个消费者收到消息均一致】

P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
C：消费者，消息的接受者，会一直等待消息到来。
Queue：消息队列，接收消息、缓存消息。
Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

exchange类型：

Fanout：广播将消息交给所有绑定到交换机的队列, 不处理路由键。只需要简单的将队列绑定到
交换机上。fanout 类型交换机转发消息是最快的。
Direct：定向把消息交给符合指定routing key 的队列. 处理路由键。需要将一个队列绑定到交换
机上，要求该消息与一个特定的路由键完全匹配。如果一个队列绑定到该交换机上要求路由键
“dog”，则只有被标记为 “dog” 的消息才被转发，不会转发 dog.puppy，也不会转发 dog.guard，
只会转发dog。
其中，路由模式使用的是 direct 类型的交换机。
Topic：主题(通配符) 把消息交给符合routing pattern（路由模式）的队列. 将路由键和某模式进
行匹配。此时队列需要绑定要一个模式上。符号 “#” 匹配一个或多个词，符号""匹配不多不少一个
词。因此“audit.#” 能够匹配到“audit.irs.corporate”，但是“audit.” 只会匹配到 “audit.irs”。
其中，主题模式(通配符模式)使用的是 topic 类型的交换机。

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

3.3.1 RabbitMQ管理界面操作

创建两个队列fanout_queue1和fanout_queue2

创建Exchange交换器fanout_exchange

将创建的fanout_exchange交换器和 fanout_queue1, fanout_queue2队列绑定

3.3.2 代码编辑

生产者（rabbitmq-producer）代码：

/**
* 目标：将消息发送给交换机，通过交换机广播给消息队列，路由键为空字符串
*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class Demo03TestPublishAndSubscribe {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@Test
	public void contextLoads() {
		//向广播交换机发送1千条消息
		for (int i = 0; i < 1000; i++) {
			/**
			* 参数1：交换机名称
			* 参数2：路由键
			* 参数3：消息内容
			*/
			rabbitTemplate.convertAndSend("fanout_exchange","","hello 我是小
			兔子【"+i+"】！");
		}
	}
}

消费者1（rabbitmq-consumer）代码：

/**
* 发布订阅模式：消息监听1
*/
@Component
@RabbitListener(queues = "fanout_queue1")
public class PubAndSubListener1 {
@RabbitHandler
public void pubAndSubHandler(String msg){
		System.out.println("=====发布订阅模式接收消息端【1】=====>"+msg);
	}
}

消费者2（rabbitmq-consumer）代码：

/**
* 发布订阅模式：消息监听2
*/
@Component
@RabbitListener(queues = "fanout_queue2")
public class PubAndSubListener2 {
	@RabbitHandler
	public void pubAndSubHandler(String msg){
		System.out.println("=====发布订阅模式接收消息端【2】=====>"+msg);
	}
}

3.4 Routing路由模式

简介：路由模式特点：队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）消息的发送方在向 Exchange发送消息时，也必须指定消息的RoutingKey。Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的Routing key完全一致，才会接收到消息。

【图解】
P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列
C1：消费者，其所在队列指定了需要routing key 为 error 的消息
C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息
【有选择性的接收消息】

3.4.1 RabbitMQ管理界面操作

创建两个队列分别叫做 routing_queue1 和 routing_queue2 用户演示

创建交换器 routing_exchange , 类型为 direct , 用于演示路由模式

设置绑定: 将创建的交换器 routing_exchange 和 routing_queue1 , routing_queue2 绑定在一起, 路由键Routing Key分别为 info 和 error

3.4.2 代码编辑

生产者（rabbitmq-producer）代码如下:

/**
* 目标：将消息发送给交换机，通过交换机路由给指定的消息队列，通过路由键类指定
*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class Demo04TestRoutingModel {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@Test
	public void contextLoads() {
		//向路由交换机发送1千条消息
		for (int i = 0; i < 1000; i++) {
			/**
			* 参数1：交换机名称
			* 参数2：路由键:error , info ，指定要投递的消息队列
			* 参数3：消息内容
			*/
			if(i%2 == 0){
				rabbitTemplate.convertAndSend("routing_exchange","info","hello 我是小兔子【"+i+"】！");
			} else {
				rabbitTemplate.convertAndSend("routing_exchange","error","hello 我是小兔子【"+i+"】！");
			}
		}
	}
}

消费者1（rabbitmq-consumer）代码如下:

/**
* 路由模式：消息队列接收监听器1，接收来自路由模式发送的消息
*/
@Component
@RabbitListener(queues = "routing_queue1")
public class RoutingListener1 {
	@RabbitHandler
	public void routingHandler(String msg){
		System.out.println("=====路由模式消息接收监听器【1】=====>"+msg);
	}
}

消费者2（rabbitmq-consumer）代码如下:

/**
* 路由模式：消息队列接收监听器2，接收来自路由模式发送的消息
*/
@Component
@RabbitListener(queues = "routing_queue2")
public class RoutingListener2 {
	@RabbitHandler
	public void routingHandler(String msg){
		System.out.println("=====路由模式消息接收监听器【2】=====>"+msg);
	}
}

3.5 Topics通配符模式(主题模式)

简介：
Topic 类型与 Direct 相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key的时候使用通配符！

Routingkey : 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：
#：匹配一个或多个词，多个词用点号分隔
❉：匹配不多不少恰好1个词
举例：
item.#：能够匹配 item.insert.abc.bbc 或者 item.insert
item.*：只能匹配 item.insert

3.5.1 RabbitMQ管理界面操作

创建队列 topic_queue1 和 topic_queue1

创建交换器 topic_exchange , type类型为 topic

设置绑定:
topic_queue1 绑定的Routing Key路由键为 item.*
topic_queue2 绑定的Routing Key路由键为 item.#

3.5.2 代码编辑

生产者（rabbitmq-producer）代码如下:

/**
* 目标：将消息发送给交换机，通过交换机路由给指定的消息队列，路由键使用通配符
*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class Demo05TestTopicModel {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@Test
	public void contextLoads() {
		//向通配符交换机发送消息
		rabbitTemplate.convertAndSend("topic_exchange","item.insert","hello
		我是小兔子，路由键item.insert");
		rabbitTemplate.convertAndSend("topic_exchange","item.insert.abc","hello 我
		是小兔子，路由键：item.insert.abc");
	}
}

消费者1（rabbitmq-producer）代码如下:

/**
* 通配符模式：消息队列接收监听器1，接收来自通配符模式发送的消息
*
*/
@Component
@RabbitListener(queues = "topic_queue1")
public class TopicListener1 {
	@RabbitHandler
	public void topicHandler(String msg){
		System.out.println("=====通配符模式消息接收监听器【1】=====>"+msg);
	}
}

消费者2（rabbitmq-producer）代码如下:

/**
* 通配符模式：消息队列接收监听器1，接收来自通配符模式发送的消息
*
*/
@Component
@RabbitListener(queues = "topic_queue2")
public class TopicListener2 {
	@RabbitHandler
	public void topicHandler(String msg){
		System.out.println("=====通配符模式消息接收监听器【2】=====>"+msg);
	}
}

总结

工作模式：
1、简单模式 HelloWorld : 一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

2、工作队列模式 Work Queue: 一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默
认的交换机）

3、发布订阅模式 Publish/subscribe: 需要设置类型为fanout的交换机，并且交换机和队列进行绑定，
当发送消息到交换机后，交换机会将消息广播发送到绑定的队列

4、路由模式 Routing: 需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing
key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

5、通配符模式 Topic: 需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式
的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

以上是关于消息队列MQ——Spring Boot整合RabbitMQ的主要内容，如果未能解决你的问题，请参考以下文章

spring boot整合activeMQ

Spring Boot：使用Rabbit MQ消息队列

SpringBoot整合RabbitMQ消息队列基本环境搭建

Rabbit MQ和Spring Boot的整合

spring boot 1.5.4 整合rabbitMQ（十七）

Spring Boot （十三）： Spring Boot 整合 RabbitMQ