SpringCloud:SpringAMQP介绍

article/2023/6/4 14:24:42

Spring AMQP是基于RabbitMQ封装的一套模板,并且还利用SpringBoot对其实现了自动装配,使用起来非常方便。Spring AMQP官方地址

在这里插入图片描述

在这里插入图片描述

Spring AMQP提供了三个功能:

  • 自动声明队列、交换机及其绑定关系
  • 基于注解的监听器模式,异步接收消息
  • 封装了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.1.111 # 主机名port: 5672 # 端口virtual-host: / # 虚拟主机username: dcxuexi # 用户名password: 123456 # 密码

然后在publisher服务中编写测试类SpringAmqpTest,并利用RabbitTemplate实现消息发送:

package com.dcxuexi.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 {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic 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.1.111 # 主机名port: 5672 # 端口virtual-host: / # 虚拟主机username: dcxuexi # 用户名password: 123456 # 密码

然后在consumer服务的com.dcxuexi.mq.listener包中新建一个类SpringRabbitListener,代码如下:

package com.dcxuexi.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来控制消费者预取的消息数量

3.发布/订阅

发布订阅的模型如图:

在这里插入图片描述

可以看到,在订阅模型中,多了一个exchange角色,而且过程略有变化:

  • Publisher:生产者,也就是要发送消息的程序,但是不再发送到队列中,而是发给交换机
  • Exchange:交换机。一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。Exchange有以下3种类型:
    • Fanout:广播,将消息交给所有绑定到交换机的队列
    • Direct:定向,把消息交给符合指定routing key的队列
    • Topic:通配符,把消息交给符合routing pattern(路由模式) 的队列
  • Consumer:消费者,与以前一样,订阅队列,没有变化
  • Queue:消息队列也与以前一样,接收消息、缓存消息。

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

4.Fanout

Fanout,英文翻译是扇出,我觉得在MQ中叫广播更合适。

在这里插入图片描述

在广播模式下,消息发送流程是这样的:

  • 1) 可以有多个队列
  • 2) 每个队列都要绑定到Exchange(交换机)
  • 3) 生产者发送的消息,只能发送到交换机,交换机来决定要发给哪个队列,生产者无法决定
  • 4) 交换机把消息发送给绑定过的所有队列
  • 5) 订阅队列的消费者都能拿到消息

我们的计划是这样的:

  • 创建一个交换机dcxuexi.fanout,类型是Fanout
  • 创建两个队列fanout.queue1fanout.queue2,绑定到交换机dcxuexi.fanout

在这里插入图片描述

4.1.声明队列和交换机

Spring提供了一个接口Exchange,来表示所有不同类型的交换机:

在这里插入图片描述

consumer中创建一个类,声明队列和交换机:

package com.dcxuexi.mq.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class FanoutConfig {/*** 声明交换机* @return Fanout类型交换机*/@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("dcxuexi.fanout");}/*** 第1个队列*/@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2个队列*/@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}

4.2.消息发送

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

@Test
public void testFanoutExchange() {// 队列名称String exchangeName = "dcxuexi.fanout";// 消息String message = "hello, everyone!";rabbitTemplate.convertAndSend(exchangeName, "", message);
}

4.3.消息接收

consumer服务的SpringRabbitListener中添加两个方法,作为消费者:

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {System.out.println("消费者1接收到Fanout消息:【" + msg + "】");
}@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {System.out.println("消费者2接收到Fanout消息:【" + msg + "】");
}

4.4.总结

交换机的作用是什么?

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

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

  • Queue
  • FanoutExchange
  • Binding

5.Direct

Fanout模式中,一条消息,会被所有订阅的队列都消费。但是,在某些场景下,我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange

在这里插入图片描述

Direct模型下:

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

案例需求如下

  1. 利用@RabbitListener声明ExchangeQueueRoutingKey

  2. consumer服务中,编写两个消费者方法,分别监听direct.queue1direct.queue2

  3. publisher中编写测试方法,向dcxuexi.direct发送消息

在这里插入图片描述

5.1.基于注解声明队列和交换机

基于@Bean的方式声明队列和交换机比较麻烦,Spring还提供了基于注解方式来声明。

consumerSpringRabbitListener中添加两个消费者,同时基于注解来声明队列和交换机:

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"),exchange = @Exchange(name = "dcxuexi.direct", type = ExchangeTypes.DIRECT),key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){System.out.println("消费者接收到direct.queue1的消息:【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"),exchange = @Exchange(name = "dcxuexi.direct", type = ExchangeTypes.DIRECT),key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){System.out.println("消费者接收到direct.queue2的消息:【" + msg + "】");
}

5.2.消息发送

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

@Test
public void testSendDirectExchange() {// 交换机名称String exchangeName = "dcxuexi.direct";// 消息String message = "红色警报!日本乱排核废水,导致海洋生物变异,惊现哥斯拉!";// 发送消息rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

5.3.总结

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

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

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

  • @Queue
  • @Exchange

6.Topic

6.1.说明

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

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

通配符规则:

#:匹配一个或多个词

*:匹配不多不少恰好1个词

举例:

item.#:能够匹配item.spu.insert 或者 item.spu

item.*:只能匹配item.spu

图示:

在这里插入图片描述

解释:

  • Queue1:绑定的是china.# ,因此凡是以 china.开头的routing key 都会被匹配到。包括china.newschina.weather
  • Queue2:绑定的是#.news ,因此凡是以 .news结尾的 routing key 都会被匹配。包括china.newsjapan.news

案例需求:

实现思路如下:

  1. 并利用@RabbitListener声明ExchangeQueueRoutingKey

  2. consumer服务中,编写两个消费者方法,分别监听topic.queue1topic.queue2

  3. publisher中编写测试方法,向dcxuexi.topic发送消息

在这里插入图片描述

6.2.消息发送

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

/*** topicExchange*/
@Test
public void testSendTopicExchange() {// 交换机名称String exchangeName = "dcxuexi.topic";// 消息String message = "喜报!胜!";// 发送消息rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

6.3.消息接收

consumer服务的SpringRabbitListener中添加方法:

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue1"),exchange = @Exchange(name = "dcxuexi.topic", type = ExchangeTypes.TOPIC),key = "china.#"
))
public void listenTopicQueue1(String msg){System.out.println("消费者接收到topic.queue1的消息:【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue2"),exchange = @Exchange(name = "dcxuexi.topic", type = ExchangeTypes.TOPIC),key = "#.news"
))
public void listenTopicQueue2(String msg){System.out.println("消费者接收到topic.queue2的消息:【" + msg + "】");
}

6.4.总结

描述下Direct交换机与Topic交换机的差异?

  • Topic交换机接收的消息RoutingKey必须是多个单词,以 **.** 分割
  • Topic交换机与队列绑定时的bindingKey可以指定通配符
  • #:代表0个或多个词
  • *:代表1个词

7.消息转换器

之前说过,Spring会把你发送的消息序列化为字节发送给MQ,接收消息的时候,还会把字节反序列化为Java对象。

在这里插入图片描述

只不过,默认情况下Spring采用的序列化方式是JDK序列化。众所周知,JDK序列化存在下列问题:

  • 数据体积过大
  • 有安全漏洞
  • 可读性差

7.1.测试默认转换器

我们修改消息发送的代码,发送一个Map对象:

@Test
public void testSendMap() throws InterruptedException {// 准备消息Map<String,Object> msg = new HashMap<>();msg.put("name", "Jack");msg.put("age", 21);// 发送消息rabbitTemplate.convertAndSend("simple.queue","", msg);
}

停止consumer服务

发送消息后查看控制台:

在这里插入图片描述

7.2.配置JSON转换器

显然,JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高,因此可以使用JSON方式来做序列化和反序列化。

publisherconsumer两个服务中都引入依赖:

<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>

配置消息转换器。

在启动类中添加一个Bean即可:

@Bean
public MessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter();
}
http://www.ngui.cc/article/show-1007633.html

相关文章

【致敬未来的攻城狮计划】连续打卡第4天+物联网操作系统概述

开启攻城狮的成长之旅&#xff01;这是我参与的由 CSDN博客专家 架构师李肯&#xff08;http://yyds.recan-li.cn&#xff09;和 瑞萨MCU &#xff08;https://www.renesas.cn/cn/zh&#xff09; 联合发起的「 致敬未来的攻城狮计划 」的第 4 天&#xff0c;点击查看活动计划详…

hastcat

hashcat 下载地址: https://hashcat.net/hashcat/ 案例 Usage: hashcat [options]... hash|hashfile|hccapxfile [dictionary|mask|directory]...https://xz.aliyun.com/t/4008破解linux shadow /etc/shadow中密码格式: $id$salt$encrypted如:$1$2eWq10AC$NaQqalCk3 1表…

ssm+vue在线课程培训系统java

在线培训系统的需求和管理上的不断提升&#xff0c;在线培训系统的潜力将无限扩大&#xff0c;在线培训系统在业界被广泛关注&#xff0c;本网站及对此进行总体分析&#xff0c;将在线培训系统信息管理的发展提供学校的管理帮助更大。 根据现有的模块&#xff0c;除管理员对系统…

docker安装Redis高可用(一主二从三哨兵)

本次教程使用docker swarm安装 准备三台机器 hostIP用途node1192.168.31.130redis-master01&#xff0c;redis哨兵节点01node2192.168.31.131redis-slave01, redis哨兵节点02node3192.168.31.132redis-slave02 redis哨兵节点02 注意事项&#xff1a; 1&#xff1a;需要保证三…

(数字图像处理MATLAB+Python)第二章数字图像处理基础-第二节:色度学基础与颜色模型

文章目录一&#xff1a;颜色匹配二&#xff1a;CIE 1931-RGB系统三&#xff1a;CIE 1931标准色度系统四&#xff1a;CIE 1976Lab均匀颜色空间五&#xff1a;孟塞尔表色系统&#xff08;1&#xff09;孟塞尔明度(Value&#xff0c;记为V)&#xff08;2&#xff09;孟塞尔彩度(Ch…

Linq使用where sql in (...)

List<int?> 机构编号数组 new List<int?>();var begin new DateTime(year, 1, 1);var end begin.AddYears(1);IQueryable<DtoReport健康体检表> Q_健康体检表 es.jktjEntitys.Where(m > m.体检日期 > begin && m.体检日期 < end).Whe…

linux 产生随机数 并遍历

1、产生随机数 varRANDOMvarRANDOM varRANDOMvar[ $var % 150 ] 2、产生不重复的随机数 $ entries($(shuf -i 0-149 -n 15)) $ echo “${entries[]}” 3、对随机数排序 $ entries($(shuf -i 0-149 -n 15 | sort -n)) $ echo “entries[]"12224549546678798393118119124140…

Vue的:key用法及示例

在 Vue 中&#xff0c;使用 v-for 指令循环遍历数组或对象时&#xff0c;每个子元素都需要绑定一个唯一的 key 值&#xff0c;用于标识这个子元素。我们可以使用 :key 属性来绑定这个 key 值。例如&#xff1a; <div v-for"item in itemList" :key"item.id&…

开发过程中遇到的关于flask的相关问题(flask入门教程,flask CORS跨域访问、base64解码为图像、PIL、CV2和bytes之间格式转换)

1、flask入门上手教程 我主要目的是快速开发&#xff0c;因此学习flask是随便找的一个教程跟着做了下&#xff0c;而没有系统地学习&#xff0c;大概差不多了就去按自己需求进行开发了。 以下是我跟的一个简单教程&#xff0c;讲述了如何快速部署一个简单的python服务器&…

“全球首款旗舰”填补行业空白,两轮电动车技术创新为何只看绿源?

作者 | 曾响铃 文 | 响铃说 乒乓作为我们的“国球”&#xff0c;在数不清的体育赛事里书写辉煌战绩&#xff0c;也进一步被国人熟知、热爱。更难能可贵的是“国球”精神&#xff1a;“别人可能练了一千次&#xff0c;而我们却练了一万次”&#xff0c;冠军品质&#xff0c;奋…