RabbitMQ 基本使用方法详解

🕗 发布于 2024-12-13 06:30 rabbitmq java 后端 spring

RabbitMQ 基本使用方法

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

1. 引入依赖

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

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

该依赖会自动导入 Spring AMQP 库以及 RabbitMQ 的客户端，允许你在 Spring 环境下方便地使用 RabbitMQ。

2. 配置 RabbitMQ

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

package com.easylive.entity.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {
 
 @Bean
    public MessageConverter messageConverter() {
        // 使用自定义的消息转换器来更严格地处理反序列化
        return new Jackson2JsonMessageConverter();
    }

    // 队列定义
    @Bean
    public Queue transferFileQueue() {
        return new Queue("transferFileRouting", true); // durable 确保队列持久化
    }

    @Bean
    public Queue videoPlayQueue() {
        return new Queue("videoPlayRouting", true);
    }

    // Direct 类型交换机
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("directExchange", true, false); // durable，是否持久化
    }

    // 队列和交换机的绑定
    @Bean
    public Binding transferFileBinding(Queue transferFileQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(transferFileQueue).to(directExchange).with("transferFileRoutingKey");
    }

    @Bean
    public Binding videoPlayBinding(Queue videoPlayQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(videoPlayQueue).to(directExchange).with("videoPlayRoutingKey");
    }
}

2.1 定义队列

在 RabbitMQ 中，队列用于存储消息，直到消费者从队列中取出。队列是消息传递的基础。

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

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

2.2 定义交换机

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

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

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

2.3 队列与交换机的绑定

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

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

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

3. 消息发送

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

for (VideoInfoFilePost filePost : addFileList) {
    amqpTemplate.convertAndSend("directExchange", "transferFileRoutingKey", filePost);
}

amqpTemplate.convertAndSend("directExchange", "videoPlayRoutingKey", videoPlayInfoDto);

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

4. 消息接收

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

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

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

5. 消息转换

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

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

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，可以方便地进行消息的序列化和反序列化。整个流程中，队列、交换机和消息的绑定机制是核心，保证了消息的有效传递和处理。

原文地址：https://blog.csdn.net/2301_80190170/article/details/144410462

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：2024年12月12日Github流行趋势
下一篇：基于双目视差的3D立体视觉原理及其应用

.NET(C#) 如何配置用户首选项及保存用户设置
.NET(C#) 如何配置用户首选项及保存用户设置
阅读更多2024-12-14
【最新】北大数字普惠金融指数数据集-省市县（2011-2023年）
郭峰,王靖一,王芳,孔涛,张勋,程志云.测度中国数字普惠金融发展:指数编制与空间特征[J].经济学(季刊),2020,19(04):1401-1418.时间跨度：省级和城市级指数时间跨度为2011-2
阅读更多2024-12-14
GESP202412 四级【Recamán】题解（AC）
a11ak−1−kkakak−1−kak−1k小杨想知道 Recamán 数列的前n项从小到大排序后的结果。手动计算非常困难，小杨希望你能帮他解决这个问题。
阅读更多2024-12-14
IDEA遇到EasyConnect中的网络资源无法访问的问题
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。原文链接：https://blog.csdn.net/wanshanyu_/article/de
阅读更多2024-12-14
双目摄像头标定方法
此时已经完成标定，左下角为反投影误差，右边为外参可视化。将双目左右目拍的图像上传（左右目最好不少于20张）此时回到主页面，即可看到成功导出。把这些误差大的删除即可。
阅读更多2024-12-14
Servlet、omcat服务器架构与工作原理
Servlet是运行在服务器端的Java程序，它的主要职责之一是接收并处理来自客户端（如浏览器）的HTTP请求。当客户端发送一个请求到服务器时，Servlet可以解析请求中的信息，例如请求的URL路径
阅读更多2024-12-14
Vue生命周期钩子函数：深入解析与实践
作为高级Vue前端开发人员，对Vue组件的生命周期钩子函数有着深刻的理解是至关重要的。生命周期钩子函数是指在Vue组件的创建、更新、销毁等过程中，Vue自动调用的一系列方法。通过这些钩子函数，我们可以
阅读更多2024-12-14
安卓开发--使用android studio发布APP
app发布
阅读更多2024-12-14
数据结构与算法学习笔记----拓扑排序
@ author: 明月清了个风。
阅读更多2024-12-14
python 将数据保存到现有的Excel文件的新工作表
out_file = ‘query.xlsx’df1 = pd.DataFrame(out_data)若直接写入：df1.to_excel(out_file, index=False, sheet_n
阅读更多2024-12-14