MQ学习笔记--(SpringAMQP)

SpringAMQP–基本介绍

什么是SpringAMQP

SpringAmqp的官方地址：https://spring.io/projects/spring-amqp

AMQP：是用于在应用程序或之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求

Spring AMQP：Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。包含两部分，其中spring-ampq是基础抽象，spring-rabbit是底层的默认实现。

SpringAMQP–入门案例的消息发送

案例：利用SpringAMQP实现HelloWorld中的基础消息队列功能

流程如下：
1.在父工程中引入spring-ampq的依赖
2.在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列
3.在consumer服务张编写消费逻辑，绑定simple.queue这个队列

步骤：步骤1：引入AMQP依赖

因为publisher和consumer服务都需要amqp依赖，因此这里把依赖直接放到父工程mq-demo中:

<!--AMPQ依赖,包含RabbitMQ-->
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1
2
3
4
5

步骤2：在publisher中编写测试方法，向simple.queue发送消息

1.在publisher服务中编写application.yaml，添加mq连接信息：

spring:
  rabbitmq:
    host: 192.168.150.101 # 主机名
    port: 5672  # 端口
    virtual-host: / # 虚拟主机
    username: luxifa # 用户名
    password: 1234 # 密码
1
2
3
4
5
6
7

2.在publisher服务中新建一个测试类，编写测试方法：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmpqTest {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@Test
	public void testSimpleQueue() {
		String queueName = "simple.queue";
		String message = "hello,spring amqp"};
		rabbitTemplate.convertAndSend(queueName, message);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12

小结：

什么是AMQP？

• 应用间消息通信的一种协议，与语言和平台无关。

SpringAMQP如何发送消息？

• 引入amqp的starter依赖？
• 配置Rabbit MQ地址
• 利用RabbitTemplate的convertAndSend方法

SpringAMQP–入门案例的消息接收

步骤3：在consumer中编写消费逻辑，监听simple.queue

1.在consumer服务中编写application.yaml，添加mq连接信息：

pring:
  rabbitmq:
    host: 192.168.150.101 # 主机名
    port: 5672  # 端口
    virtual-host: / # 虚拟主机
    username: luxifa # 用户名
    password: 1234 # 密码
1
2
3
4
5
6
7

2.在consumer服务中新建一个类，编写消费逻辑

@Component
public class SpringRabbitListener {
	
	@RabbitListener(queues = "simple.queue")
	public void listenSimpleQueueMessage(String msg) throws InterruptedException {
		System.out.println("spring 消费者接收到消息: [" + msg + "]");
	}
}
1
2
3
4
5
6
7
8

小结：

SpringAMQP如何接收消息？

• 引入amqp的starter依赖
• 配置RabbitMQ地址
• 定义类，添加@Component注解
• 类中声明方法，添加@RabbitListener注解，方法参数就是消息

注意：消息一旦消费就会从队列删除，RabitMQ没有消息回溯功能

SpringAMQP–WorkQueue模型

Work Queue 工作队列

Work queue，工作队列，可以提高消息处理速度，避免队列消息堆积

publisher --> queue  -->多个consumer
1

案例：模拟WoekQueue，实现一个队列绑定多个消费者

基本思路如下：
1.在publisher服务中定义测试方法，每秒产生50条消息，发送到simple.queue
2.在consumer服务中定义两个消息监听者，都监听simple.queue队列
3.消费者1每秒处理50条消息，消费者2每秒处理10条消息

步骤 步骤1：生产者循环发送消息到simple.queue

在publisher服务中添加一个测试方法，循环发送50条消息到simple.queue队列

@Test
pubic 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(200);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13

步骤 步骤2：编写两个消费者，都监听simple.queue

在consumer服务中添加一个消费者，也监听simple.queue

@RabbitListener(queue = "simple.queue")
public void listenSimpleQueueMessage1(String msg) throws InterruptedException {
	System.out.println("spring 消费者1接收到消息：[" + msg + "]");
	Thread.sleep(25);
}

@RabbitListener(queue = "simple.queue")
public void listenSimpleQueueMessage2(String msg) throws InterruptedException {
	System.out.println("spring 消费者2接收到消息：[" + msg + "]");
	Thread.sleep(100);
}
1
2
3
4
5
6
7
8
9
10
11

我们期望的是能者多劳，第一个方法能多处理些消息，实际结果确实两者处理消息的个数是一样的。

消费预存限制

修改application.yml文件，设置preFetch这个值，可以控制预取消息的上限：

spring:
  rabbitmq:
    host: 192.168.150.101 # 主机名
    port: 5672 # 端口
    virtual-host: / #虚拟主机
    username: luxifa # 用户名
    password: 1234 # 密码
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息,处理完成才能获取下一个消息
1
2
3
4
5
6
7
8
9
10

加上这个之后，方法1处理的消息比方法2多

小结

Work模型的作用

• 多个消费者绑定到同一个队列，同一条消息只会被一个消费者处理
• 通过设置prefetch来控制消费者预存的消息数量

SpringAMQP–发布订阅模型介绍

发布(Publish)、订阅(Subscribe)

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实际方法是加入了exchage(交换机)。

常见exchange类型包括：

• Fanout：广播
• Direct：路由
• Topic：话题

publisher-->exchange-->多个queue-->多个consumer
1

注意：exchange负责消息路由，而不是存储，路由失败则消息丢失

SpringAMQP–FanoutExchange

发布订阅-Fanout Exchange

Fanout Exchange会将收到的消息路由到每一个跟其绑定的queue

案例 利用SpringAMQP演示FanoutExchange的使用

实现思路如下：
1.在consumer服务中，利用代码申明队列、交换机，并将两者绑定
2.在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2
3.在publisher中编写测试方法，向luxifa.fanout发送消息

步骤 步骤1：在consumer服务声明Exchange、Queue、Binding

在consumer服务常见一个类，添加@Configuration注解，并声明FanoutExchage、Queue和绑定关系对象Binding，代码如下

@Configuration
public class FanoutConfig {
	// 声明FanoutExchange交换机
	@Bean
	public FanoutExchange fanoutExchage(){
		return new FanoutExchange("luxifa.fanout");
	}
	// 申明第一个队列
	@Bean
	public Queue fanoutQueue1(){
		return new Queue("fanout.queue1");
	}
	// 绑定队列1和交换机
	//Queue fanoutQueue1 这个地方的名称fanoutQueue1不能写错，按照类型和名称注入
	@Bean
	public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
		return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
	}

	//以相同方式申明第2个队列，并完成绑定
	// 申明第二个队列
	@Bean
	public Queue fanoutQueue2(){
		return new Queue("fanout.queue2");
	}
	// 绑定队列2和交换机
	//Queue fanoutQueue2 这个地方的名称fanoutQueue2不能写错，按照类型和名称注入
	@Bean
	public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
		return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

步骤 步骤2：在consumer服务声明两个消费者

在consumer服务的SpringRabbitListener类中，添加两个方法，分别监听fanout.queue1和fanout.queue2：

@RabbitListener(queue = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
	System.out.println("消费者接收到fanout.queue1的消息：[" + msg + "]");
}
@RabbitListener(queue = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
	System.out.println("消费者接收到fanout.queue2的消息：[" + msg + "]");
}
1
2
3
4
5
6
7
8

步骤 步骤3：在publisher服务发送消息到FanoutExchange

在publisher服务的SpringAmqpTest类中添加测试方法：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmpqTest {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@Test
	public void testSendFanoutExchange() {
	    // 交换机名称
		String exchangeName= "luxifa.fanout";
		// 消息
		String message = "hello";
		//发送消息,参数分别是：交换机名称、RoutingKey(暂时未空)、消息
		rabbitTemplate.convertAndSend(exchangeName, routingKey:"", message);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

小结

交换机的作用是什么？

• 接收publisher发送的消息
• 将消息按照规则路由到与之绑定的队列
• 不能缓存消息，路由失败，消息丢失
• FanoutExchange的会将消息路由到每个绑定的队列

声明队列、交换机、绑定关系的Bean是什么？

• Queue
• FanoutExchange
• Binding

SpringAMQP–DirectExchange

Direct Exchange会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式(routes)。

• 每一个Queue都与Exchange设置一个BindingKey
• 发布者发送消息时，指定消息的RoutingKey
• Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

publisher-->Direct Exchange key:blue-->假设两个queue,queuq1的bindingkey：blue  queue2的bindingkey：yellow --> 每个queue都有一个consumer
1

queuq1和queuq2 也可以指定相同的bindingkey，比如都是bindingkey:red

案例：利用ApringAMQP演示DirectExchange的使用

实现思路如下：
1.利用@RabbitListener声明Exchange、Queue、RoutingKey

publisher-->exchange luxifa:direct -->两个queue,一个叫direct.queue1 bindingKey：blue、red  一个叫direct.queue2  bindingKey：yellow、red  --> 每个queue都有一个consumer
1

步骤：步骤1 在consumer服务声明Exchange、Queue

1. 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
3. 并利用@RabbitListener声明Exchange、Queue、RoutingKey

@RabbitListener(binding = @QueueBinding(
		value = @Queue(name = "direct.queue1"),
		exchange = @Exchange(name = "luxifa.direct", type = "Exchangetypes.DIRECT"),
		key = {"red","bule"}
))
public void listenDirectQueue1(String msg) {
	System.out.println("消费者1接收搭配Direct消息：[" +msg+ "]");
}

@RabbitListener(binding = @QueueBinding(
		value = @Queue(name = "direct.queue2"),
		exchange = @Exchange(name = "luxifa.direct", type = "Exchangetypes.DIRECT"),
		key = {"red","yellow"}
))
public void listenDirectQueue1(String msg) {
	System.out.println("消费者2接收搭配Direct消息：[" +msg+ "]");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

步骤：步骤2 在publisher服务发送消息到DirectExchange

在publisher服务的SpringAmqpTest类中添加测试方法：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmpqTest {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@Test
	public void testSendDirectExchange() {
	    // 交换机名称
		String exchangeName= "luxifa.direct";
		// 消息
		String message = "hello, blue";
		//发送消息,参数分别是：交换机名称、RoutingKey(暂时未空)、消息
		rabbitTemplate.convertAndSend(exchangeName, "blue", message);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

小结

描述下Direct交换机与Fanout交换机的差异？

• Fanout交换机将消息路由给每一个与之绑定的队列
• Direct交换机根据RoutingKey判断路由给哪个队列
• 如果多个队列具有相同的RoutingKey，则与Fanout功能类似

基于@RabbitListener注解声明队列与交换机有哪些常见注解？

• @Queue
• @Exchange

SpringAMQP–TopicExchange

发布订阅-TopicExchange

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以 .分割
Queue与Exchange指定BindingKey时可以使用通配符：

• #：代指0个或多个单词
• *：代指一个单词

比如：

• china.news 代表有中国的新闻消息；
• china.weather代表中国的天气消息；
• japan.news 代表日本新闻
• japan.weather 代表日本的天气消息；

publisher-->exchange topic -->四个queue,一个叫queue1 bindingKey：china.#  一个叫queue2  bindingKey：japan.# 一个叫queue3 bindingKey：#.weather  一个叫queue4  bindingKey：#.news  --> 每个queue都有一个consumer
1

案例：利用SpringAMQP演示TopicExchange的使用

实现思路如下：
1.利用@RabbitListener声明Exchange、Queue、RoutingKey
2.在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
3.在publisher中编写测试方法，向luxifa.topic发送消息

步骤：步骤1 在consumer服务声明Exchange、Queue

1.在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
2.利用@RabbitListener声明Exchange、Queue、RoutingKey

publisher-->exchange luxifa.topic -->二个queue,一个叫topic.queue1 bindingKey：china.#  一个叫topic.queue2  bindingKey：#.news  --> 每个queue都有一个consumer
1

@RabbitListener(binding = @QueueBinding(
		value = @Queue(name = "topic.queue1"),
		exchange = @Exchange(name = "luxifa.topic", type = "Exchangetypes.TOPIC"),
		key = "china.#"
))
public void listenTopicQueue1(String msg) {
	System.out.println("消费者1接收到topic.queue1消息：[" +msg+ "]");
}

@RabbitListener(binding = @QueueBinding(
		value = @Queue(name = "topic.queue2"),
		exchange = @Exchange(name = "luxifa.topic", type = "Exchangetypes.TOPIC"),
		key = "#.news"
))
public void listenTopicQueue2(String msg) {
	System.out.println("消费者2接收到topic.queue2消息：[" +msg+ "]");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

步骤：步骤2 在publisher服务发送消息到TopicExchange

在publisher服务的SpringAmqpTest类中添加测试方法：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmpqTest {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@Test
	public void testSendTopicExchange() {
	    // 交换机名称
		String exchangeName= "luxifa.topic";
		// 消息
		String message = "hello";
		//发送消息,参数分别是：交换机名称、RoutingKey(暂时未空)、消息
		rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

SpringAMQP–消息转换器

案例：测试发送Object类型消息

说明：在SpringAMQP的发送消息中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列化为字节后发送。

我们在consumer中利用@Bean声明一个队列

@Bean
public Queue objectMessageQueue() {
	return new Queue("object.queuq")}
1
2
3

在publisher中发送消息以测试：

@Test
public void testSendMap() throws InterruptException {
	// 准备消息
	Map<String,Object> msg = new HashMap<>();
	msg.put("name","jack");
	msg.put("age",21);
	//发送消息
	rabbitTemplate.convertAndSend("object.queue",msg);
}
1
2
3
4
5
6
7
8
9

消息转换器

Spring的消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理。而默认实现是SimpleMessageConvertes，基于JDK的ObjectOutputStream完成序列化。
如果要修改只需要定义一个MessageConverter类型的Bean即可。推荐使用JSON方式序列化，步骤如下：

在publisher服务引入依赖

<dependency>
	<groupId>com.fasterxml.jackson.dataformat</groupId>
	<artifactId>jackson-dataformat-xml</artifactId>
	<version>2.9.10</version>
</dependency>
1
2
3
4
5

我们在publisher服务声明MessageConverter

@Bean
public MessageConverter jsonMessageConverter(){
	return new Jackson2JsonMessageConverter();
}
1
2
3
4

我们在consumer服务中引入jackson依赖：

<dependency>
	<groupId>com.fasterxml.jackson.dataformat</groupId>
	<artifactId>jackson-dataformat-xml</artifactId>
	<version>2.9.10</version>
</dependency>
1
2
3
4
5

我们在consumer服务定义MessageConverter：

@Bean
public MessageConverter jsonMessageConverter(){
	return new Jackson2JsonMessageConverter();
}
1
2
3
4

然后定义一个消费者，监听obejct.queue队列并消费消息：

@RabbitListener(queues = "object.queue")
public void ListenObjectQueue(Map<String, Object> msg) {
	System.out.println("收到消息：【" + msg +"】");
}
1
2
3
4

小结

SpringAMQP中消息的序列化和反序列化是怎么实现的？

• 利用MessageConverter实现的，默认是JDK的序列化
• 注意发送方与接收方必须使用相同的MessageConverter