spring配置MQ自动创建队列和exchange

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了spring配置MQ自动创建队列和exchange相关的知识,希望对你有一定的参考价值。

参考技术A Rabbit MQ加上@Configuration配置,在mq上没有队列和exhange的时候,自动创建队列和exchange

消息队列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将消息发送到对应的队列

以上是关于spring配置MQ自动创建队列和exchange的主要内容,如果未能解决你的问题,请参考以下文章

spring中配置监听队列的MQ

01 Rabbit MQ 的初步使用

SpringBoot日记——MQ消息队列整合

简单讲解RabbitMQ

RabbitMQ基础总结

RabbitMQ基础总结