SpringCloud-AMQP

news/2024/5/5 15:22:11

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

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

image

SpringAMQP提供了三个功能：

  • 自动声明队列、交换机及其绑定关系
  • 基于注解的监听器模式，异步接收消息
  • 封装了RabbitTemplate工具，用于发送消息

父工程pom.xml

<!--AMQP依赖，包含RabbitMQ-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

application.yml

spring:rabbitmq:host: 192.168.88.130port: 5672username: lmcoolpassword: 1234virtual-host: /
Appdata:rabbitMQ-name: demo-mq

Basic Queue 简单队列模型

publisher.sendMessage2demomq
利用RabbitTemplate实现消息发送

package com.lmcode.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.beans.factory.annotation.Value;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;@SpringBootTest
@RunWith(SpringRunner.class)
public class SpringAMQPTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Value("${Appdata.rabbitMQ-name}")private String mqname;@Testpublic void sendMessage2demomq(){String message = "hello springamqp";rabbitTemplate.convertAndSend(mqname,message);}
}

consumer.listenSimpleQueueMessage
消费者监听消息

package com.lmcode.mq.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class SpringRabbitListener {@RabbitListener(queues = "demo-mq")public void listenSimpleQueueMessage(String msg) throws InterruptedException {System.out.println("成功接收消息---【" + msg + "】");}
}

Work Queue 工作队列模型

image

消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。

Work queues，也被称为（Task queues），任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。

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

consumer.application.yml
控制预取消息的上限，根据消费者的能力处理消息，不是平均分配

spring:rabbitmq:listener:simple:prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

publisher.sendMessage2demomqWorkQueue
向队列中不停发送消息，模拟消息堆积。

    @Testpublic void sendMessage2demomqWorkQueue() throws InterruptedException {String message = "hello, message_";for (int i = 0; i < 50; i++) {rabbitTemplate.convertAndSend(mqname, message + i);Thread.sleep(20);}}

consumer.SpringRabbitListener
线程阻塞，模拟任务耗时

    @RabbitListener(queues = "demo-mq")public void listenWorkQueueMessage1(String msg) throws InterruptedException {System.out.println("成功接收消息---【" + msg + "】---消费者1---WorkQueue" + LocalTime.now());Thread.sleep(20);}@RabbitListener(queues = "demo-mq")public void listenWorkQueueMessage2(String msg) throws InterruptedException {System.err.println("成功接收消息---【" + msg + "】---消费者2---WorkQueue" + LocalTime.now());Thread.sleep(200);}

发布、订阅模型

image

Exchange：交换机一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：

  • Fanout：广播，将消息交给所有绑定到交换机的队列
  • Direct：定向，把消息交给符合指定routing key 的队列
  • Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

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

发布、订阅模型-Fanout【扇出、广播】

image

  • 广播模式的消息发送流程：

可以有多个队列，每个队列都要绑定到Exchange（交换机），生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定，交换机把消息发送给绑定过的所有队列，订阅队列的消费者都能拿到消息

  • 交换机的作用：

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

声明队列和交换机并绑定

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

consumer.config

package com.lmcode.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 {// 定义交换机，队列，绑定队列到交换机@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("lm.fanoutExchange");}@Beanpublic Queue fanoutQueue1(){return new Queue("lm.fanoutQueue1");}@Beanpublic Queue fanoutQueue2(){return new Queue("lm.fanoutQueue2");}@Beanpublic Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}@Beanpublic Binding fanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}

image
image
image

publisher.sendMessage2demomqFanoutQueue
消息发送给交换机，参数：交换机名，路由规则，消息

    @Testpublic void sendMessage2demomqFanoutQueue(){String exchangeName = "lm.fanoutExchange";String message = "hello, message";rabbitTemplate.convertAndSend(exchangeName,"",message);}

consumer.SpringRabbitListener

    @RabbitListener(queues = "lm.fanoutQueue1")public void listenFanoutQueueMessage1(String msg) throws InterruptedException {System.out.println("成功接收消息---【" + msg + "】---消费者1---FanoutQueue" + LocalTime.now());}@RabbitListener(queues = "lm.fanoutQueue2")public void listenFanoutQueueMessage2(String msg) throws InterruptedException {System.err.println("成功接收消息---【" + msg + "】---消费者2---FanoutQueue" + LocalTime.now());}

发布、订阅模型-Direct【基于注解声明队列和交换机】

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

image

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

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

可以指定多个key，相同的key相当于Fanout，但是性能会不好

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

基于@RabbitListener注解声明队列和交换机：@QueueBinding、@Exchange、@Queue

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

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

consumer.SpringRabbitListener

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "lm.directQueue1"),exchange = @Exchange(name = "lm.directExchange", type = ExchangeTypes.DIRECT),key = {"key1", "key2"}))public void listenDirectQueue1(String msg){System.out.println("成功接收消息---【" + msg + "】---消费者1---directQueue1" + LocalTime.now());}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "lm.directQueue2"),exchange = @Exchange(name = "lm.directExchange", type = ExchangeTypes.DIRECT),key = {"key1", "key3"}))public void listenDirectQueue2(String msg){System.err.println("成功接收消息---【" + msg + "】---消费者2---directQueue2" + LocalTime.now());}

publisher.sendMessage2demomqDirectQueue

    @Testpublic void sendMessage2demomqDirectQueue(){String exchangeName = "lm.directExchange";String message1 = "hello, key1";String message2 = "hello, key2";String message3 = "hello, key3";rabbitTemplate.convertAndSend(exchangeName,"key1",message1);rabbitTemplate.convertAndSend(exchangeName,"key2",message2);rabbitTemplate.convertAndSend(exchangeName,"key3",message3);}

发布、订阅模型-Topic

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

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

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

Routingkey 一般都是有一个或多个单词组成，多个单词之间以.分割

image

解释：

  • Queue1：绑定的是china.# ，因此凡是以 china.开头的routing key 都会被匹配到。包括china.news和china.weather
  • Queue2：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配。包括china.news和japan.news

  1. 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2

  2. 在publisher中编写测试方法，向itcast. topic发送消息

image

publisher.sendMessage2demomqTopicQueue

    @Testpublic void sendMessage2demomqTopicQueue() {String exchangeName = "lm.topicExchange";String message1 = "hello, china.news";String message2 = "hello, china.weather";String message3 = "hello, japan.news";String message4 = "hello, japan.weather";rabbitTemplate.convertAndSend(exchangeName, "china.news", message1);rabbitTemplate.convertAndSend(exchangeName, "china.weather", message2);rabbitTemplate.convertAndSend(exchangeName, "japan.news", message3);rabbitTemplate.convertAndSend(exchangeName, "japan.weather", message4);}

consumer.SpringRabbitListener

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "lm.topicQueue1"),exchange = @Exchange(name = "lm.topicExchange", type = ExchangeTypes.TOPIC),key = "china.#"))public void listenTopicQueue1(String msg){System.out.println("成功接收消息---【" + msg + "】---消费者1---topicQueue1" + LocalTime.now());}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "lm.topicQueue2"),exchange = @Exchange(name = "lm.topicExchange", type = ExchangeTypes.TOPIC),key = "japan.#"))public void listenTopicQueue2(String msg){System.err.println("成功接收消息---【" + msg + "】---消费者2---topicQueue2" + LocalTime.now());}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "lm.topicQueue4"),exchange = @Exchange(name = "lm.topicExchange", type = ExchangeTypes.TOPIC),key = "#.news"))public void listenTopicQueue3(String msg){System.out.println("成功接收消息---【" + msg + "】---消费者3---topicQueue3" + LocalTime.now());}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "lm.topicQueue4"),exchange = @Exchange(name = "lm.topicExchange", type = ExchangeTypes.TOPIC),key = "#.weather"))public void listenTopicQueue4(String msg){System.err.println("成功接收消息---【" + msg + "】---消费者4---topicQueue4" + LocalTime.now());}

消息转化器【配置JSON转换器】

Spring会把发送的消息序列化为字节发送给MQ，接收消息时把字节反序列化为Java对象；Spring对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。

使用java JDK的序列化方式：

  • 可读性差
  • 可能出现注入问题，有安全漏洞
  • 数据体积过大，传输消息速度慢，而且额外占用内存空间，性能差
@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);
}

发送消息后查看控制台：

image

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

如果要修改只需要定义一个MessageConverter类型的Bean即可。

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

consumer/publisher.pom.xml

<dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId>
</dependency>

consumer/publisher.Application.MessageConverter
配置消息转换器

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

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hjln.cn/news/24801.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈，一经查实，立即删除！

相关文章

Pandas read_csv 参数详解

Pandas read_csv 参数详解

前言 在使用 Pandas 进行数据分析和处理时，read_csv 是一个非常常用的函数，用于从 CSV 文件中读取数据并将其转换成 DataFrame 对象。read_csv 函数具有多个参数，可以根据不同的需求进行灵活的配置。本文将详细介绍 read_csv 函数的各个参数及其用法，帮助大家更好地理解和利…
阅读更多...
（数据科学学习手札160）使用miniforge代替miniconda

（数据科学学习手札160）使用miniforge代替miniconda

本文已收录至我的Github仓库https://github.com/CNFeffery/DataScienceStudyNotes1 简介大家好我是费老师，conda作为Python数据科学领域的常用软件，是对Python环境及相关依赖进行管理的经典工具，通常集成在anaconda或miniconda等产品中供用户日常使用。但长久以来，conda在很…
阅读更多...
SHCG_Numerical_results

SHCG_Numerical_results

阅读更多...
Caused by: java.lang.ClassNotFoundException: org.junit.runner.manipulation.Filter问题的解决

Caused by: java.lang.ClassNotFoundException: org.junit.runner.manipulation.Filter问题的解决

问题描述问题解决 后来经过查阅资料发现，是这里出现了问题：只需要将JUnit的路径更改到classpath下面就可以啦：问题完美解决！
阅读更多...
games101-3 BRDF101

games101-3 BRDF101

BRDF101 概述 本文基于知乎Maple对brdf的文章，在此基础又收集了一些其它来源的关于brdf的文章，希望能够完全理解记忆相关知识 关于Jakub Boksansky的文章，看的过程中又去搜集了很多其它文章来理解，发现已经超出了我目前的知识厚度，因此只会简单的翻译一下我能理解的部分，…
阅读更多...
开源相机管理库Aravis例程学习（四）——multiple-acquisition-signal

开源相机管理库Aravis例程学习（四）——multiple-acquisition-signal

本文针对Aravis官方例程中的：02-multiple-acquisition-signal做简单的讲解，并介绍其中部分函数目录简介例程代码函数说明g_main_loop_newg_main_loop_rung_main_loop_quitg_signal_connectarv_stream_set_emit_signalsQ&A回调函数的同步调用与异步调用帧丢失问题 简介 本…
阅读更多...
openCV 图像清晰度检测

openCV 图像清晰度检测

图像清晰度评价算法有很多种，在空域中，主要思路是考察图像的领域对比度，即相邻像素间的灰度特征的梯度差；在频域中，主要思路是考察图像的频率分量，对焦清晰的图像高频分量较多，对焦模糊的图像低频分量较多。 这里实现3种清晰度评价方法，分别是Tenengrad梯度方法、Lapla…
阅读更多...
Linux：VMware切换仅主机模式并配置静态IP

Linux：VMware切换仅主机模式并配置静态IP

配置网络编辑器 点击“编辑”->“虚拟网络编辑器”没有仅主机模式的话，可以通过“添加网络”进行新增网络配置。更改虚拟机网路模式 右键“创建的虚拟就”->“设置”登录虚拟机配置静态IP 切换目录到“/etc/sysconfig/network-scripts/”修改“if-ens33”文件TYPE=Ether…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023