程序员经验分享-hwhale

【SpringCloud】SpringAMQP总结

2023-11-20

1、AMQP

Advanced Message Queuing Protocol，高级消息队列协议。是用于在应用程序之间传递业务消息的开放标准。

该协议与语言和平台无关，更符合微服务中独立性的要求。

在这里插入图片描述

Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。

它包含两部分，其中spring-amqp是基础抽象，spring-rabbit是底层的默认实现。

在这里插入图片描述

接下来对几种常见的MQ模型，用SpringAMQP来演示一下具体实现。

2、基本消息模型队列

消息发送

  • 引入AMQP依赖（publisher和consumer服务都需要amqp依赖，因此这里把依赖直接放到父工程）
<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>    		
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
  • 在publisher服务中编写application.yml，添加mq连接信息
spring:
  rabbitmq:
      host: 192.168.150.101 # 主机名
      port: 5672 # 端口
      virtual-host: / # 虚拟主机 
      username: root # 用户名
      password: 123321 # 密码
  • 在publisher服务中新建一个测试类，注入RabbitTemplate，调用convertAndSend方法，传入队列名称和消息内容即可。
@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);
    }
}

消息接收

  • 在consumer服务中编写application.yml，添加mq连接信息
spring:
  rabbitmq:
      host: 192.168.150.101 # 主机名
      port: 5672 # 端口
      virtual-host: / # 虚拟主机 
      username: root # 用户名
      password: 123321 # 密码
  • 在consumer服务中新建一个类，编写消费逻辑
//定义类，添加@Component注解
@Component
public class SpringRabbitListener {
	@RabbitListener(queues = "simple.queue")    
	public void listenSimpleQueueMessage(String msg) throws InterruptedException {        
		System.out.println("spring 消费者接收到消息 ：【" + msg + "】");    
	}
}

类中声明方法，添加@RabbitListener注解，注解上写明要消费的队列名称，此时，方法的参数就是消息。

在这里插入图片描述

3、WorkQueue模型

和基本模型不一样，WorkQueue模型，即工作队列，有多个消费者。可以提高消息处理速度，避免队列消息堆积。

在这里插入图片描述

案例：实现一个队列绑定多个消费者

思路如下：

  • 在publisher服务中定义测试方法，每秒产生50条消息，发送到队列simple.queue
  • 在consumer服务中定义两个消息监听者，都监听simple.queue队列，但消费者1每秒能处理50条消息，而消费者2每秒只能处理10条消息

演示代码如下，先写生产者：

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

@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);    
     }
}

接下来编写两个消费者，都监听上面的队列simple.queue：

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

@RabbitListener(queues = "simple.queue") 
public void listenSimpleQueueMessage2(String msg) throws InterruptedException {    
	System.err.println("spring 消费者2接收到消息：【" + msg + "】");    
	Thread.sleep(100);

}

上面一个用System.out.println一个用System.error.println，这样一红一白，在控制台查看消费的效果：

在这里插入图片描述

可以看到，消费者2处理消息慢（代码里用处理一次休眠的旧来模拟），但它从队列里拿的消息却和消费者1一样，因此导致消费总时间过长。对消费者2来说，这就是没有那个金刚钻，却揽了这么多瓷器活儿。这个现象的原因是 ⇒ 消息预取机制

预取机制，通俗说就是消息到了队列后，消费者通过通道一人一个先拿过来，能不能快速处理完的另说，先取走再说。

接下来加一个消费预取限制。通过设置prefetch来控制消费者预取的消息数量。修改消费者的application.yml文件，设置preFetch这个值，可以控制预取消息的上限：

spring:
  rabbitmq:
      host: 192.168.150.101 # 主机名
      port: 5672 # 端口
      virtual-host: / # 虚拟主机 
      username: root # 用户名
      password: 123321 # 密码
	  listener:      
	  	simple:        
	  	  prefetch: 1  # 每次只能获取一条消息，处理完成才能获取下一个消息

prefetch默认无限，这里改为1，即 每次只能获取一条消息，处理完成才能获取下一个消息。重启，这次同样生产50条消息，消费总时间变短了，可以看到消费者2处理的慢，取的也慢。

在这里插入图片描述

4、发布订阅模型

和前面两种模型不同，一个消息被一个消费者消费完就没了。发布订阅模式允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。 当前上一章说的支付功能的实现，就得用这个模型。

在这里插入图片描述

消息被路由到哪些队列中，由exchange决定。常见exchange类型包括：

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

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

5、发布订阅-Fanout Exchange

Fanout Exchange 会将接收到的消息广播到每一个跟其绑定的queue

在这里插入图片描述

SpringAMQP提供了声明交换机、队列、绑定关系的API：

在这里插入图片描述

接下来写演示代码：

  • 步骤1：在消费consumer服务声明Exchange、Queue，并Binding绑定
//在consumer服务配置目录下创建一个类，添加@Configuration注解
@Configuration
public class FanoutConfig {
    // 声明FanoutExchange交换机
    @Bean    
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("exchange.fanout");
    }    
    // 声明第1个队列
    @Bean    
    public Queue fanoutQueue1(){        
    	return new Queue("fanout.queue1");    
	}   

	//绑定队列1和交换机，方法形参就是队列和交换机类型的参数  
	@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和交换机  
	@Bean    
	public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){        
		return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);    
	}    
}

//将来Spring读到这些Bean，就会向RabbitMQ去声明这些队列和交换机，并绑定

启动消费服务，可以看到队列创建并且绑定到了交换机，以后交换机有消息，和交换机绑定的队列都能收到一份！

在这里插入图片描述

  • 步骤2：在consumer服务声明两个消费者，即添加两个方法，分别监听fanout.queue1和fanout.queue2 
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
   System.out.println("消费者1接收到fanout.queue1的消息：【" + msg + "】");
}

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

//这里在一个服务里用两个消费者方法模拟被多个服务消费
//到此，两个队列就分别和这两个消费者方法勾搭上了
//重启消费者服务。
  • 步骤3：在publisher服务发送消息到FanoutExchange
//这次不再是send到队列，而是发到交换机
@Test
public void testFanoutExchange() {    // 队列名称    String exchangeName = "itcast.fanout";    // 消息    String message = "hello, everyone!";    // 发送消息，参数分别是：交互机名称、RoutingKey（暂时为空）、消息     rabbitTemplate.convertAndSend(exchangeName, "", message);}

//执行这段测试代码，往交换机发消息

查看消费者服务控制台，可以看到一次发送，被多个消费者收到

在这里插入图片描述

小总结：

交换机的作用是什么？

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

6、发布订阅-DirectExchange

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

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

在这里插入图片描述

当然两个队列可以设置同一个key，此时的效果就和上面的Fanout Exchange一样了。

在这里插入图片描述

步骤1：在consumer服务声明Exchange、Queue。这里不再用Bean，而是利用@RabbitListener声明Exchange、Queue、RoutingKey

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


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

看IDEA的提示类型写就行：exchange的type属性默认就是direct类型，这里不加也行。

在这里插入图片描述
启动消费者服务，到此，队列和交换机都被创建和绑定好了，查看RabbitMQ控制台：

在这里插入图片描述

步骤2：在publisher服务发送消息到DirectExchange，此时convertAndSend方法的第二个参数就要写Routing key了

//在publisher服务的SpringAmqpTest类中添加测试方法
@Test
public void testDirectExchange() {    
	// 队列名称    
	String exchangeName = "itcast.direct";    
	// 消息     
	String message = "hello,blue!";    
	// 发送消息，参数依次为：交换机名称，RoutingKey，消息
	 rabbitTemplate.convertAndSend(exchangeName, "blue", message);
}

运行这个测试方法，查看消费服务：

在这里插入图片描述

小总结：

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

基于@RabbitListener注解声明队列和交换机的常用注解有：
@Queue
@Exchange

7、发布订阅-TopicExchange

TopicExchange与DirectExchange类似，区别在于它的routingKey必须是多个单词的列表，并且以 . 分割

在这里插入图片描述

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

举例：

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

总之就是用通配符简化了队列和exchange的绑定。用代码演示一下这个TopicExchange类型的使用，测试结构：

在这里插入图片描述
步骤1： 在consumer服务利用@RabbitListener声明Exchange、Queue、RoutingKey

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

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

//注意key和exchange中的type

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

//在publisher服务的SpringAmqpTest类中添加测试方法
@Test
public void testTopicExchange() {
    // 队列名称    
    String exchangeName = "itcast.topic";    
    // 消息     
    String message = "喜报！孙悟空大战哥斯拉，胜!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message);

}

8、消息转换器

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

//这次不再是用@RabbitListener，因为它用在消费方法上，我们要看消息，不希望它被消费
//因此使用Bean声明队列
@Bean
public Queue objectMessageQueue(){
    return new Queue("object.queue");
}

在publisher中发送消息以测试：

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

重启服务，查看RabbitMQ中的消息：

在这里插入图片描述
数据展示有问题，Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。接下来修改序列化的方式。

修改序列化方式只需要定义一个MessageConverter 类型的Bean即可

  • 在父工程（或者消息生产者服务）中引入依赖
<dependency>
   <groupId>com.fasterxml.jackson.core</groupId> 
   <artifactId>jackson-databind</artifactId>
</dependency>
  • 在配置类中声明MessageConverter的Bean（直接写启动类中也行）
//我们一旦声明MessageConverter，就会覆盖Spring默认的MessageConverter（SpringBoot自动装配的原理）

@Bean
public MessageConverter jsonMessageConverter(){
    return new Jackson2JsonMessageConverter(); 
}

此时再重发消息，RabbitMQ中消息展示正常。接下来看消费者服务：

  • 引入Jackson依赖（如果上一步引入依赖放在了父工程，那这里就不用重复引入了）
<dependency>
   <groupId>com.fasterxml.jackson.core</groupId> 
   <artifactId>jackson-databind</artifactId>
</dependency>
  • 在consumer服务定义MessageConverter：
@Bean
public MessageConverter jsonMessageConverter(){
    return new Jackson2JsonMessageConverter(); 
}
  • 定义一个消费者，监听object.queue队列并消费消息
@RabbitListener(queues = "object.queue")
public void listenObjectQueue(Map<String, Object> msg) {
    System.out.println("收到消息：【" + msg + "】"); 
}

重启消费者服务，序列化方式修改成功！

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

利用MessageConverter实现的，默认是JDK的序列化，可重新定义MessageConverter类型的Bean来修改。

注意发送方与接收方必须使用相同的MessageConverter

System.out.println("End!");
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

springCloud

spring cloud

Java

javarabbitmq

【SpringCloud】SpringAMQP总结 的相关文章

随机推荐

  • Python Logging 模块

    Python Logging 模块 1 Logging 模块介绍 介绍连接 https www jianshu com p 7b5e4752932e Python 中的 logging 模块可以让你跟踪代码运行时的事件 当程序崩溃时可以查看

  • 刷新完固件后opkg update报错的解决方法

    刷新完固件后opkg update报错的解决方法 一 更改设备ip 当你使用lan口接入局域网后 如果你ping不通局域网上的其他设备ip 那么需要更改ip vim etc config network 修改lan口的ip为局域网下同一网段

  • js中相对路径写法

    表示上一级目录开始 表示同级目录开始 表示根目录开始

  • display和visibility的区别

  • 微信小程序-日期时间选择器

    我在官方表单组件的基础上改出了几种比较常用的日期 时间 选择器 为了更清晰 我没给任何样式 wxss可根据个人喜好添加 一 picker view 嵌入页面的滚动选择器 注意 其中只可放置组件 其他节点不会显示 picker view co

  • 1.7 聪明的投资者： 积极型投资者的证券组合策略-被动的方法

    积极型投资者 指出不应该做哪些事情 不购买高等级的优先股 低等级债券 如果一公司要出很高的利息才能借到钱 说明该公司风险较高 要购买高等级的应税债券和优质的免税债券 二级债券和优先股 信用等级较低的公司很难卖出普通债券 不可转换债券 所以他

  • springboot自定义favicon.ico不生效

    本人以下均实验过都不成功 最后成功的办法为 在 resources static favicon ico 更换浏览器为谷歌浏览器 下面是其他方法 关闭默认图标 在application properties中添加 spring mvc fa

  • Docker进阶学习: swarm扩展学习

    swarm 集群的管理和编排 docker可初始化一个 swarm 集群 其他节点可加入 manager worker Node 就是一个docker节点 多个节点组成了一个网络集群 manager worker Service 服务 可管

  • 交易的本质 什么样的信仰，决定什么样的交易人生

    什么样的信仰 决定什么样的生活 同样 什么样的理念 也决定了什么样的交易 多数的交易员都把能否在交易市场稳定化盈利归结于自己从事这个行业的时间 通常很多人会说五年入门 十年小成等等 好像只要坚持个五年十年就能够在交易市场找到自己的位置 你信

  • JAVA WEB 中间件为SERVLET(四)

    写一个用户登录部署到tomcat 本地 先找到一个模板 HTML代码复制到本地的项目index jsp中 这个登录模板包含一个JSP 一个JS 三个CSS等文件 这个是index jsp代码

  • 解决git commit 报错WARNING: Block comments use a trailing */ on a separate line

    首先用一段报错的code作为示例 struct ctrl opt dtg pt IP DTG 0xf8 0x01 IP DTG 0xf0 0x00 git commit会报错 WARNING Block comments should al

  • 办公小技巧：VSCode无法安装插件怎么办？

    Visual Studio Code是Microsoft开发的一款开源免费的现代化轻量级代码编辑器 它体积小 启动快 系统内存占用率低非常适合搭建IDE VSCode不仅支持几乎所有主流的开发语言的语法高亮 智能代码补全 自定义热键 括号匹

  • 解决 qml module “QtQuick“ is not installed问题

    在编写QWidget QML 程序上 编译通过运行没有问题 但是点击加载qml窗口时 软件运行时提示module QtQuick Window is not installed 这个错误问题 其实是缺少于是把qt 安装目录下的qml模块相关

  • C语言 命名和关键字

    变量的命名规则 变量名可以由字母 数字和 下划线 组合而成 变量名不能包含除 以外的任何特殊字符 如 逗号 空格等 变量名必须以字母或 下划线 开头 变量名不能包含空白字符 换行符 空格和制表符称为空白字符 C 语言中的某些词 例如 int

  • SSH框架相关准备与入门学习

    最近开始学习java web开发 记录一下学习的过程 主要分为三个步骤 1 基础 java Mysql入门学习 2 中级 html css javascipt servlet jsp入门学习 推荐韩顺平相关视频 w3cshool在线教程 一

  • golang sleep

    golang的休眠可以使用time包中的sleep 函数原型为 func Sleep d Duration 其中的Duration定义为 type Duration int64 Duration的单位为 nanosecond 为了便于使用

  • python——selenium

    一 Selenium Python环境搭建及配置 1 1 selenium 介绍 selenium 是一个 web 的自动化测试工具 不少学习功能自动化的同学开始首选 selenium 因为它相比 QTP 有诸多有点 免费 也不用再为破解

  • cpolar内网穿透+ EasyImage组合，自建一个图床网站

    文章目录 1 前言 2 EasyImage网站搭建 2 1 EasyImage下载和安装 2 2 EasyImage网页测试 2 3 cpolar的安装和注册 3 本地网页发布 3 1 Cpolar云端设置 3 2 Cpolar内网穿透本地

  • 【马士兵】Python基础--15

    Python基础 15 文章目录 Python基础 15 编程思想 类与对象 类的创建 对象的创建 类属性 类方法 静态方法 动态绑定属性和方法 知识点总结 编程思想 类与对象 python中一切皆对象 类的创建 类的名称由一个或多个单词组

  • 【SpringCloud】SpringAMQP总结

    文章目录 1 AMQP 2 基本消息模型队列 3 WorkQueue模型 4 发布订阅模型 5 发布订阅 Fanout Exchange 6 发布订阅 DirectExchange 7 发布订阅 TopicExchange 8 消息转换器

热门标签