RabbitMQ 根本使用方法详解

RabbitMQ 根本使用方法

在你的代码中，涉及到了 RabbitMQ 的根本使用，包括队列定义、交换机的配置、消息的发送与接收等内容。下面我将详细总结 RabbitMQ 的根本使用方法，重点解释如何在 Spring Boot 项目中与 RabbitMQ 集成。
1. 引入依赖

在 Spring Boot 项目中，使用 Spring AMQP 组件来集成 RabbitMQ。首先需要在 pom.xml 中添加相关的依赖：

1. <dependency>
2.     <groupId>org.springframework.boot</groupId>
3.     <artifactId>spring-boot-starter-amqp</artifactId>
4. </dependency>
5. <dependency>
6.     <groupId>com.fasterxml.jackson.core</groupId>
7.     <artifactId>jackson-databind</artifactId>
8. </dependency>

复制代码

该依赖会主动导入 Spring AMQP 库以及 RabbitMQ 的客户端，答应你在 Spring 环境下方便地使用 RabbitMQ。
2. 配置 RabbitMQ

首先，你需要配置 RabbitMQ 的相关参数（如连接信息、交换机、队列等）。在你的例子中，RabbitConfig 类就负责了这些配置。

1. package com.easylive.entity.config;
3. import org.springframework.amqp.core.Binding;
4. import org.springframework.amqp.core.BindingBuilder;
5. import org.springframework.amqp.core.DirectExchange;
6. import org.springframework.amqp.core.Queue;
7. import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
8. import org.springframework.amqp.support.converter.MessageConverter;
9. import org.springframework.context.annotation.Bean;
10. import org.springframework.context.annotation.Configuration;
12. @Configuration
13. public class RabbitConfig {
14.        
15.         @Bean
16.     public MessageConverter messageConverter() {
17.         // 使用自定义的消息转换器来更严格地处理反序列化
18.         return new Jackson2JsonMessageConverter();
19.     }
21.     // 队列定义
22.     @Bean
23.     public Queue transferFileQueue() {
24.         return new Queue("transferFileRouting", true); // durable 确保队列持久化
25.     }
27.     @Bean
28.     public Queue videoPlayQueue() {
29.         return new Queue("videoPlayRouting", true);
30.     }
32.     // Direct 类型交换机
33.     @Bean
34.     public DirectExchange directExchange() {
35.         return new DirectExchange("directExchange", true, false); // durable，是否持久化
36.     }
38.     // 队列和交换机的绑定
39.     @Bean
40.     public Binding transferFileBinding(Queue transferFileQueue, DirectExchange directExchange) {
41.         return BindingBuilder.bind(transferFileQueue).to(directExchange).with("transferFileRoutingKey");
42.     }
44.     @Bean
45.     public Binding videoPlayBinding(Queue videoPlayQueue, DirectExchange directExchange) {
46.         return BindingBuilder.bind(videoPlayQueue).to(directExchange).with("videoPlayRoutingKey");
47.     }
48. }

复制代码

2.1 定义队列

在 RabbitMQ 中，队列用于存储消息，直到消耗者从队列中取出。队列是消息转达的根本。

1. @Bean
2. public Queue transferFileQueue() {
3.     return new Queue("transferFileRouting", true); // durable 确保队列持久化
4. }

复制代码

• Queue("transferFileRouting", true)：创建一个名为 transferFileRouting 的队列，true 表示该队列是恒久化的，即 RabbitMQ 会在服务器重启后保存队列。
  

2.2 定义交换机

交换机（Exchange）负责接收来自生产者的消息，并根据队列绑定的规则将消息路由到相应的队列。RabbitMQ 支持差异类型的交换机（如 Direct, Fanout, Topic 等），在这个例子中使用的是 Direct Exchange。

1. @Bean
2. public DirectExchange directExchange() {
3.     return new DirectExchange("directExchange", true, false); // durable，是否持久化
4. }

复制代码

• DirectExchange("directExchange", true, false)：创建一个名为 directExchange 的交换机，true 表示该交换机是恒久化的，false 表示不主动删除。
  

2.3 队列与交换机的绑定

队列和交换机之间的绑定决定了消息如何路由。在你的例子中，队列通过 Routing Key 和交换机进行绑定。

1. @Bean
2. public Binding transferFileBinding(Queue transferFileQueue, DirectExchange directExchange) {
3.     return BindingBuilder.bind(transferFileQueue).to(directExchange).with("transferFileRoutingKey");
4. }

复制代码

• BindingBuilder.bind(transferFileQueue).to(directExchange).with("transferFileRoutingKey")：这表示将 transferFileQueue 队列与 directExchange 交换机进行绑定，使用 transferFileRoutingKey 作为路由键。
  

3. 消息发送

消息生产者通过交换机发送消息到队列，消耗者从队列中获取消息并处置惩罚。你在代码中的生产者部门使用了 amqpTemplate.convertAndSend 方法来发送消息，消息是发送给direct交换机，并指定key值。

1. for (VideoInfoFilePost filePost : addFileList) {
2.     amqpTemplate.convertAndSend("directExchange", "transferFileRoutingKey", filePost);
3. }
5. amqpTemplate.convertAndSend("directExchange", "videoPlayRoutingKey", videoPlayInfoDto);

复制代码

• amqpTemplate.convertAndSend("directExchange", "transferFileRoutingKey", filePost)：这表示通过交换机 directExchange，使用 transferFileRoutingKey 作为路由键，发送 filePost 对象到消息队列。
  
• amqpTemplate.convertAndSend 会主动序列化对象（这里使用 Jackson2JsonMessageConverter）并发送到队列。
  

4. 消息接收

消耗者使用 @RabbitListener 注解监听队列中的消息。当队列中有消息时，消耗者会触发相应的处置惩罚方法。

1. @RabbitListener(queues = "transferFileRouting")
2. public void consumeTransferFileQueue(@Payload VideoInfoFilePost videoInfoFile) {
3.     try {
4.         videoInfoPostService.transferVideoFile(videoInfoFile);
5.     } catch (Exception e) {
6.         log.error("处理转码文件队列消息失败", e);
7.     }
8. }

复制代码

• @RabbitListener(queues = "transferFileRouting")：表示该方法会监听名为 transferFileRouting 的队列。
  
• @Payload 注解用于指定消息体的类型。这里接收的是 VideoInfoFilePost 类型的消息。
  
• 消耗者方法会在消息到达时被触发，执行相应的业务逻辑。
  

5. 消息转换

Spring AMQP 提供了 MessageConverter 接口，可以用于消息内容的转换。在你的配置中，使用了 Jackson2JsonMessageConverter 作为消息转换器，它会将消息对象转换为 JSON 格式发送，并在接收时进行反序列化。

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

复制代码

6. 完整流程总结

• 队列和交换机的定义与配置：
  
  
  
  • 定义队列和交换机，设置恒久化属性。
    
  • 使用 Binding 将队列与交换机绑定，并指定路由键。
    
  
  
• 消息生产者发送消息：
  
  
  
  • 使用 amqpTemplate.convertAndSend 发送消息到指定的交换机，并根据路由键将消息发送到特定的队列。
    
  • 发送的消息会经过 MessageConverter 转换成 JSON 格式。
    
  
  
• 消息消耗者接收消息：
  
  
  
  • 使用 @RabbitListener 注解来监听队列中的消息。
    
  • 当消息到达时，消耗者会主动触发相应的方法，并处置惩罚消息。
    
  
  
• 异常处置惩罚：
  
  
  
  • 消耗者中可以加入异常处置惩罚逻辑（如 try-catch），以确保在消息处置惩罚失败时记录日记并进行得当的处置惩罚。
    
  
  

7. 注意事项

• 恒久化与确认机制：
  
  
  
  • 如果消息队列和交换机是恒久化的，那么即使 RabbitMQ 重启，队列和交换机也会被保存。但这要求队列中的消息也需要恒久化，否则消息会丢失。
    
  
  
• 事务与确认：
  
  
  
  • Spring AMQP 提供了事务支持，可以在发送消息时确保消息的可靠性。
    
  • 消耗者可以使用 @RabbitListener 的 ackMode 属性来控制消息确认机制，包管消息被成功消耗后才从队列中移除。
    
  
  
• 消息格式与序列化：
  
  
  
  • 使用 Jackson2JsonMessageConverter 作为默认消息转换器可以方便地进行 Java 对象与 JSON 格式的互转。
    
  
  
• 死信队列与重试机制：
  
  
  
  • RabbitMQ 支持死信队列（DLX）和消息重试机制，用于处置惩罚消耗失败的消息。
    
  
  

总结

RabbitMQ 在 Spring Boot 中的集成非常简便，通过 @Configuration 配置类定义队列、交换机和绑定关系，生产者通过 amqpTemplate 发送消息，消耗者使用 @RabbitListener 监听队列消息。结合 MessageConverter，可以方便地进行消息的序列化和反序列化。整个流程中，队列、交换机和消息的绑定机制是核心，包管了消息的有用转达和处置惩罚。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。