0
0

【RabbitMQ（二）】：Exchange 详解 | Message Convert 消息转换器

文章目录

03. 使用 Java 代码去操控 RabbitMQ

💡 Spring 提供了对 RabbitMQ 的支持：Spring AMQP


💡 

AMQP

和 

Spring AMQP
  • AMQP（Advanced Message Queuing Protocol）：是用在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。
  • Spring AMQP：是 Spring 基于 AMQP 设计的一套 API 规范，提供了 发送和接收消息 的 API。 - 声明式配置：Spring AMQP 提供了一组注解和 XML 配置来声明和配置消息队列、交换机、绑定等 AMQP 元素，使得配置变得简单和灵活。- 消息发送和接收：Spring AMQP 提供了 AmqpTemplate 接口，用于发送和接收消息。开发者可以使用 AmqpTemplate 接口中的方法来发送消息到队列或交换机，并从队列或交换机中接收消息。- 消息监听容器：Spring AMQP 提供了 @RabbitListener 注解和 SimpleMessageListenerContainer 类来简化消息监听的配置和管理。开发者可以使用 @RabbitListener 注解将方法标记为消息监听器，并通过配置 SimpleMessageListenerContainer 来管理消息监听器的运行。- 消息转换：Spring AMQP 提供了一组消息转换器（MessageConverter），用于在消息发送和接收之间进行消息 格式 的转换。开发者可以使用消息转换器来将消息对象转换为字节流，或将字节流转换为消息对象，以实现消息的序列化和反序列化。- 事务管理：Spring AMQP 提供了对事务的支持，开发者可以使用 RabbitTransactionManager 类来管理消息发送和接收的事务，以确保消息的可靠传递。

3.1 快速入门

3.1.1 创建父子项目

💡 这里使用 maven 创建父子项目来演示，父项目命名为 

mq-demo

，下面包含两个子项目 

publisher

和 

consumer

来分别扮演消息发送者和消息消费者。

👉 创建父项目

👉 依赖中选择 

Lombok

Spring Web

Spring for RabbitMQ

👉 删除掉不需要的部分

👉 更改 

pom.xml

文件中的打包方式为 

pom

，且需要注册 

modules

<packaging>pom</packaging><modules><module>publisher</module><module>consumer</module></modules>

👉 创建子项目 

publisher

，无需选择依赖

👉 修改其父项目为 

mq-demo

<parent><groupId>com.example</groupId><artifactId>mq-demo</artifactId><version>0.0.1-SNAPSHOT</version></parent>

👉 以相同的方法创建 

consumer

Module

👉 打开 maven 的 

Group Modules

选项

👉 最终父项目配置文件参考

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.2.2</version><relativePath/><!-- lookup parent from repository --></parent><groupId>com.example</groupId><artifactId>mq-demo</artifactId><version>0.0.1-SNAPSHOT</version><name>mq-demo</name><modules><module>publisher</module><module>consumer</module></modules><packaging>pom</packaging><description>mq-demo</description><properties><java.version>21</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency><!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-core --><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-core</artifactId><version>2.15.2</version></dependency><dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></exclude></excludes></configuration></plugin></plugins></build></project>
3.1.2 编写代码

💡 这里使用将消息添加到队列和监听队列的方式去演示

上面父项目中已经导入了相关的包，子项目可以直接引用

👉 在 

application.yaml

中配置 

RabbitMQ

服务器

spring:rabbitmq:host: 192.168.88.3 # 服务器的 ip 地址port:5672# 服务器端口virtual-host: /myHost # 主机名username: root # 账号password:123456# 密码

👉 创建一个测试使用的队列

👉 将 

publisher

的代码写在 

test

文件中，首先要引入 

RabbitTemplate

，这是对请求代码的 高级封装。

@SpringBootTestclassPublisherApplicationTests{@ResourceRabbitTemplate rabbitTemplate;@TestvoidcontextLoads(){String queueName ="simple.queue";String message ="Hello RabbitMq!";
        rabbitTemplate.convertAndSend(queueName, message);}}

👉 

consumer

中的代码写在 

/listeners/MqListener

中，使用 

@RabbitListener

来指定监听的队列

packagecom.example.consumer.listeners;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@Slf4j@ComponentpublicclassMqListener{@RabbitListener(queues ="simple.queue")publicvoidlisten(String message){
        log.info("收到消息{}", message);}}

💡 需要注意的是两个子模块的配置文件中都需要表明服务器的相关配置。

👉 运行 

consumer

和 

publisher

中的代码，检查是否成功接收到消息

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

3.2 Work 模型

💡 

RabbitMQ

的 

Work

模型是一种简单的消息队列模型，也被称为简单队列模式（

Simple Queue Pattern

）。在 

Work

模型中，多个消费者共享同一个队列，并从该队列中获取消息进行处理。

  • 这个模型有一个特点叫做 消息均衡消费，即：RabbitMQ 会将消息均匀地分配给各个消费者进行处理，以实现负载均衡。即使消费者的处理能力不同，RabbitMQ 也会尽量将消息均匀分发，以保证每个消费者都能够参与到消息的处理中。

👉 先来测试一下这个模型，在 

consumer

中创建两个 方法 去监听同一个端口，一个方法处理完后休眠 

1000ms

，另一个不休眠，统计它们处理消息的数量：

int listener1 =1;int listener2 =1;@RabbitListener(queues ="work.queue")publicvoidlistenerWorkQueue1(String msg)throwsInterruptedException{
        log.info("listener1 收到消息{}，目前共处理消息{}条", msg, listener1++);}@RabbitListener(queues ="work.queue")publicvoidlistenerWorkQueue2(String msg)throwsInterruptedException{
        log.error("listener2 收到消息{}，目前共处理消息{}条", msg, listener2++);Thread.sleep(1000);}

💡 后面不再每次强调队列的创建，大家可以自主创建队列来进行模拟或者参考我代码中的队列。

👉 利用 

publisher

发送消息

@TestvoidpublishTest()throwsInterruptedException{String queueName ="work.queue";for(int i =0; i <50; i++){String message ="Hello RabbitMq! id = "+ i;
            rabbitTemplate.convertAndSend(queueName, message);}}

👉 观察控制台，可以发现两个方法分别处理了 

25

条消息，即使它们的处理速度差别非常大。

❗ 但肯定希望能力强的 

consumer

多处理一些部分，这就需要了解 

RabbitMQ

的预取机制，在 

RabbitMQ

中，预取（

Prefetch

）是指消费者在处理消息之前，从 

RabbitMQ

服务器预取一定数量的消息到本地缓冲区的行为。预取机制允许消费者一次性从服务器获取多个消息，以提高消息处理的效率和吞吐量。

💡 

Spring AMQP

会根据消费者的数量来 平均 分配预取的数量，这就导致了上面出现的各 

25

条的现象。

👉 所以为了使得预取数量可以按照自己的能力来取得，可以手动将预取的数量设定为 

1

也就是 

consumer

处理完这一条消息后再去取下一条，这样就实现了按照工作能力来分配消息。

application.xml

spring:rabbitmq:host: 192.168.88.3
    port:5672virtual-host: /myHost # 主机名username: root
    password:123456listener:simple:prefetch:1

👉 再次测试

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

  • 处理速度快的直接就处理了 49 条消息。

3.3 RabbitMQ 中的三类交换机

💡 真实的生产活动中都是通过 

exchange

也就是交换机，来发送消息

❓为什么要通过交换机来发消息而不是直接发送到队列呢？

  • 首先是后续拓展的问题，后续如果加入一个 consumer 而它的队列又不是已有的队列，这时候就需要新增 publisher 中的代码；而通过交换机则不同 consumer 可以将自己指定的队列接入到交换机（这在 consumer 方是很容易实现的），publisher 只需要将消息发布到 exchange 就可以实现分配，实现了生产者和消费者的 解耦，提高了系统的灵活性。
  • 交换机可以 根据路由规则 将消息转发到符合条件的队列中，从而实现消息的 过滤和转发。通过合理设置交换机的绑定规则，可以实现精确的消息路由和分发，满足不同业务场景下的需求。
  • RabbitMQ 提供了不同类型的交换机，如直连交换机（Direct Exchange）、主题交换机（Topic Exchange）、扇出交换机（Fanout Exchange）等。每种交换机类型都有不同的路由规则和特点，可以根据实际需求选择合适的交换机类型来实现灵活的消息路由。
3.3.1 Fanout 扇出交换机

💡 

Fanout Exchange

会将其接收到的消息发布到每一个和其 

bind

的队列

  • 可以实现发布/订阅模式（Pub/Sub），publisher 将消息通过该交换机广播出去，所以也被称为广播模式。

👉 监听端和上面的代码相同，没什么变化，仍然指向监听的队列即可，将这两个队列绑定到 

amq.fanout

上

@RabbitListener(queues ="fanout.queue1")publicvoidlistenerFanoutQueue1(String msg)throwsInterruptedException{
        log.info("Fanout listener1 收到消息{}", msg);}@RabbitListener(queues ="fanout.queue2")publicvoidlistenerFanoutQueue2(String msg)throwsInterruptedException{
        log.error("Fanout listener2 收到消息{}", msg);}

💡 

amq.fanout

是系统创建时默认携带的交换机，当然也可以根据业务去创建自己的交换机

  • 通过在图形化控制界面指定新创建的交换机的 type 来创建不同的交换机

👉 编写 

publisher

代码

@TestvoidFanoutTest()throwsInterruptedException{String exchangeName ="amq.fanout";String message ="Hello Everyone!";
            rabbitTemplate.convertAndSend(exchangeName,"",message);Thread.sleep(20);}

💡 这里仍然使用上面的 

convertAndSent

的 API

  • 通过指定 routingKey（必须要有，没有的话以空字符串填充）和 exchangeName
  • 这里看一下两个重写方法的区别：// 交换机rabbitTemplate.convertAndSend(exchangeName,"",message);// 队列rabbitTemplate.convertAndSend(queueName, message);

👉 发送信息，查看效果

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

3.3.2 Direct 交换机

💡 显然只通过广播方式无法满足实际开发的需求

  • Direct 交换机可以根据消息的路由键（Routing Key）将消息直接发送到与之匹配的队列中。
  • Direct 的路由键就是其绑定队列时的绑定键，这与后面提到的 Topic 交换机有区别
  • Direct 交换机通常用于 点对点 的消息传递模式，即一条消息只能被一个消费者处理，当生产者发送一条消息时，Direct 交换机会根据消息的路由键将消息发送到一个特定的队列中，然后只有这个队列的一个消费者能够接收和处理该消息。

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

👉 按照上图的形式去创建队列

  • 但一般不会采用这种将一个 routing key 去绑定多个队列的场景，这里只是作为演示。

👉 编写 

publisher

的 

test

方法

@TestvoidDirectTest()throwsInterruptedException{String exchangeName ="amq.direct";String message ="Hello Everyone!";
        rabbitTemplate.convertAndSend(exchangeName,"blue",message);Thread.sleep(20);}

👉 编写接收消息的方法，仍然是只需要指定监听的队列

@RabbitListener(queues ="direct.queue1")publicvoidlistenerDirectQueue1(String msg)throwsInterruptedException{
        log.error("Direct listener1 收到消息{}", msg);}@RabbitListener(queues ="direct.queue2")publicvoidlistenerDirectQueue2(String msg)throwsInterruptedException{
        log.info("Fanout listener2 收到消息{}", msg);}@RabbitListener(queues ="direct.queue3")publicvoidlistenerDirectQueue3(String msg)throwsInterruptedException{
        log.error("Fanout listener3 收到消息{}", msg);}

👉 发送消息并且测试
在这里插入图片描述

3.3.3 Topic 交换机

💡 

Topic

交换机是 

RabbitMQ

中的一种高级交换机类型，它允许消息根据路由键（

Routing Key

）的模式匹配将消息路由到 一个或多个队列 中。Topic 交换机使用 通配符 匹配路由键，使得消息能够更灵活地进行路由和分发。

  • Topic 交换机通常用于 多对多 的消息传递模式，即一条消息可以被多个消费者处理。
  • 可以将 Topic Exchange 看作是对 Direct Exchange 的一种拓展，它的 Routing Key 可以是多个单词，中间以 . 来进行分割，比如 china.news 和 china.weather，而 Binding Key 支持 通配符。- # 代表一个或者多个单词- * 代表一个单词比如将 Binding Key 设置为 china.# 则 Routing Key 为两个中的任意一个都会导航到这个队列

👉 编写 

publisher

代码

@TestvoidTopicTest()throwsInterruptedException{String exchangeName ="amq.topic";String news ="Hello Everyone!";String weather ="晴朗";
        rabbitTemplate.convertAndSend(exchangeName,"china.weather", weather);
        rabbitTemplate.convertAndSend(exchangeName,"china.news", news);Thread.sleep(20);}

👉 编写 

consumer

代码

@RabbitListener(queues ="topic.queue1")publicvoidlistenerTopicQueue1(String msg)throwsInterruptedException{
        log.info("Topic listener1 收到消息{}", msg);}@RabbitListener(queues ="topic.queue2")publicvoidlistenerTopicQueue2(String msg)throwsInterruptedException{
        log.info("Topic listener2 收到消息{}", msg);}

👉 发送消息测试结果

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

3.4 声明队列交换机

💡 

Spring AMQP

提供了一些 API 去创建新的队列和交换机可以在代码中去创建新的交换机

  • 在生产环境（development environment）中使用的 RabbitMQ 和运行环境中使用的肯定是不同的，所以在部署的时候需要重新设置交换机和队列，而在控制台上创建这些队列就可能会导致很多由于人工出现的错误。
  • 所以可以选择在代码中去创建类和属性，这样可以避免大部分的人工问题。
3.4.1 方式一：书写 Config 类

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

💡 

Spring AMQP

提供了几个类用来声明队列、交换机以及其绑定关系

  • Queue 类：用于声明队列，可以使用 QueueBuilder 中的 build() 方法
  • Exchange 类：声明交换机
  • Binding 类：声明队列与交换机的关系

👉 通过将这些 

Bean

交给 

spring

去管理就可以实现自主的创建。

packagecom.example.consumer.config;importorg.springframework.amqp.core.*;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassFanoutConfiguration{@BeanpublicFanoutExchangefanoutExchange(){//        ExchangeBuilder.fanoutExchange("").build();returnnewFanoutExchange("fanout2");}@BeanpublicQueuefanout2Queue(){//        return QueueBuilder.durable("").build();returnnewQueue("fanout2.queue1");}@BeanpublicBindingbinding(){returnBindingBuilder.bind(fanout2Queue()).to(fanoutExchange());}}
3.4.2 方式二：注解方式创建

👉 代码示例

@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="object.queue1", declare ="true"),
            exchange =@Exchange(name ="object", type =ExchangeTypes.DIRECT),
            key ={"object"}))publicvoidlistenerObject(Map<String,String> map){System.out.println(map.get("name"));}

💡 通过这样的注解就可以声明出队列、交换机和它们之间的 

binding
  • 通过 value 指定 queue，后面的 declare 是是否持久化的选项，默认为 true（后面详细讲）
  • 通过 exchange 指定交换机，ExchangeTypes 为枚举类，指定创建的类型
  • 通过 key 来指定 Binding Key，是一个字符串数组，可以指定多个 Binding Key
  • 当使用 @RabbitListener 注解时，如果指定的队列不存在，Spring Boot 会 自动创建该队列。如果指定的交换机不存在，Spring Boot 也会 自动创建该交换机。然后将 队列与交换机进行绑定，以便消息可以正确地路由到指定的队列。

3.5 消息转换器

💡 在发送 对象 的时候 

Spring

对消息对象的处理是基于 JDK 的 

ObjectOutputStream

来序列化的

  • 这就导致了下面几个问题- JDK 序列化有安全风险- JDK 序列化后的消息体积很大- JDK 序列化的消息可读性很差

👉 所以在处理发送对象的时候使用 

JSON

格式来对信息进行序列化，首先引入依赖

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

👉 配置消息转换器到 

ApplicationContext

中，即声明 

Bean

，

  • 可以通过写 Config 类的方式实现，这里直接将声明写到主方法中
@SpringBootApplicationpublicclassPublisherApplication{publicstaticvoidmain(String[] args){SpringApplication.run(PublisherApplication.class, args);}@BeanpublicMessageConverterMessageConverter(){returnnewJackson2JsonMessageConverter();}}

@SpringBootApplicationpublicclassConsumerApplication{publicstaticvoidmain(String[] args){SpringApplication.run(ConsumerApplication.class, args);}@BeanpublicMessageConverterMessageConverter(){returnnewJackson2JsonMessageConverter();}}

❗ 注意：在 

publisher

和 

consumer

中的序列化以及反序列化的转换器 必须相同

👉 实现对象的传递

@TestvoidObjectTest()throwsJsonProcessingException{Map<String,String> map =newHashMap<>();
        map.put("name","Jack");
        map.put("age","18");String exchangeName ="object";CorrelationData correlationData =newCorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(exchangeName,"object", map);}

@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="object.queue1", declare ="true"),
            exchange =@Exchange(name ="object", type =ExchangeTypes.DIRECT),
            key ={"object"}))publicvoidlistenerObject(Map<String,String> map){System.out.println(map.get("name"));}
标签： rabbitmq ruby 分布式
本文转载自: https://blog.csdn.net/weixin_74895237/article/details/136092486
版权归原作者 *Soo_Young* 所有， 如有侵权，请联系我们删除。
登录后发布评论

“【RabbitMQ（二）】：Exchange 详解 | Message Convert 消息转换器”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航