RabbitMQ原理、使用与实践指南

🕗 发布于 2025-01-16 07:57 java-rabbitmq rabbitmq java spring boot spring cloud

一、概念

RibbitMQ 是由 erlang 语言开发，基于 AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列。
它是一种应用程序之间的通信方法，在分布式系统开发中应用广泛。

二、特点

可靠性：支持持久化、传输确认、发布确认等，保证了 MQ 的可靠性。
灵活的分发消息策略：在消息进入 MQ 前由 Exchange (交换机) 进行路由消息。
分发消息策略有：
- 简单模式；
- 工作队列模式；
- 发布订阅模式；
- 路由模式；
- 通配符模式。
支持集群：多台 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker。
多种协议：支持多种消息队列协议，如 STOMP、MQTT 等。
支持多种语言客户端：几乎支持所有常用编程语言，包括 Java、.NET、Ruby 等。
可视化管理界面：提供了一个易用的用户界面，使用户可以监控和管理消息 Broker。
插件机制：提供了许多插件，可通过插件进行扩展，也可编写自己的插件。

三、AMQP 和 JMS MQ 的区别和联系

JMS：
- 定义了统一的接口，来对消息操作进行统一；
- 限定了必须使用 Java 语言；
- 规定了两种消息模型。
AMQP：
- 通过规定协议来统一数据交互的格式；
- 不规定实现方式，因此是跨语言的；
- 消息模型更加丰富。

四、RabbitMQ 的核心组件

Broker：消息队列服务进程，此进程包括两个部分：Exchange 和 Queue。
Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过滤。
Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的。
Producer：消息生产者，即生产方客户端，生产方客户端将消息发送。
Consumer：消息消费者，即消费方客户端，接收 MQ 转发的消息。

五、RabbitMQ 的执行流程

（一）生产者发送消息流程

生产者和 Broker 建立 TCP 连接。
生产者和 Broker 建立通道。
生产者通过通道将消息发送给 Broker，由 Exchange 将消息进行转发。
Exchange 将消息转发到指定的 Queue（队列）。

（二）消费者接收消息流程

消费者和 Broker 建立 TCP 连接。
消费者和 Broker 建立通道。
消费者监听指定的 Queue（队列）。
当有消息到达 Queue 时 Broker 默认将消息推送给消费者。
消费者接收到消息。
ack（消息确认机制）回复。

六、RabbitMQ 的六种消息模式

基本消息模式
Work 消息模式（工作队列或者竞争消费者模式）
Publish/subscribe（交换机类型：Fanout，也称为广播模式）
Routing 路由模式（交换机类型：direct）
Topics 通配符模式（交换机类型：topics）
RPC 模式

七、RabbitMQ 的消息确认机制（ACK）

（一）ACK 机制类型

自动确认（Auto ACK）
- 当消费者接收消息时，RabbitMQ 自动认为消息已被成功处理，会立即将消息从队列中移除。这种方式简单，但可能会导致消息丢失，因为消费者可能在处理消息的过程中发生故障而未能真正完成处理。例如，若一个消费者接收到消息但在执行存储消息到数据库的操作时崩溃，消息将丢失，因为 RabbitMQ 已将其从队列中删除。
手动确认（Manual ACK）
- 消费者在接收到消息后，不会自动确认，需要在消息处理完成后调用 channel.basicAck 进行确认。如果在处理消息过程中消费者发生故障，RabbitMQ 会将消息重新发送给其他消费者或在该消费者恢复后重新发送。这对于重要消息的处理非常重要，确保消息不会因为消费者的临时故障而丢失。
否定确认（Nack）
- 使用 channel.basicNack 可以对消息进行否定确认，可设置是否重新入队。例如 channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); 表示拒绝消息并将其重新入队。这在处理异常情况时非常有用，如消息格式错误，可将其重新入队以便后续处理。
拒绝消息（Reject）
- 使用 channel.basicReject 可拒绝消息，但不能批量操作，如 channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false); 表示拒绝消息且不重新入队。可用于处理那些确定无法处理的消息，避免它们继续留在队列中。

（二）ACK 机制选择

如果消息不太重要，丢失也没有影响，那么自动 ACK 会比较方便。
如果消息非常重要，不容丢失。那么最好在消费完成后手动 ACK，否则接收消息后就自动 ACK，RabbitMQ 就会把消息从队列中删除。如果此时消费者宕机，那么消息就丢失了。

public class WorkQueueConsumerWithManualAck {
    private final static String QUEUE_NAME = "work_queue";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String receivedMessage = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + receivedMessage + "'");
            try {
                // 模拟消息处理可能出现的异常
                if (receivedMessage.contains("error")) {
                    throw new Exception("Message processing error");
                }
                // 这里可以添加消息处理逻辑
                System.out.println("Message processed successfully");
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            } catch (Exception e) {
                System.err.println("Message processing failed: " + e.getMessage());
                // 可以根据具体情况决定是否重新入队
                // channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
            }
        };
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });
    }
}

在上述代码中，消费者在成功处理消息后调用channel.basicAck方法确认消息。如果处理消息过程中出现异常，可以根据情况决定是否使用channel.basicNack方法将消息重新入队。

（三）注意事项

消费者数量与性能：增加消费者数量可以提高任务处理速度，但并不是消费者数量越多越好。过多的消费者可能会导致系统资源竞争（如 CPU、内存、网络带宽等），反而降低整体性能。需要根据实际的任务类型、服务器资源等因素来合理配置消费者数量。
消息顺序：由于 RabbitMQ 默认采用轮询方式分配消息，在大多数情况下可以保证消息的顺序处理。但如果消费者处理消息的速度不同，可能会导致消息的顺序在一定程度上被打乱。如果对消息顺序有严格要求，可能需要在消费者端或者消息设计上采取额外的措施来保证顺序。
消息持久化：为了防止消息丢失，尤其是在工作队列模式下多个消费者处理大量任务的情况，考虑将队列和消息进行持久化。这样在 RabbitMQ 服务重启或者出现故障时，消息不会丢失。可以通过在声明队列和发送消息时设置相应的参数来实现消息持久化。例如，在声明队列时将队列设置为持久化：

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

在发送消息时使用持久化属性：

channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));

八、RabbitMQ 如何避免消息堆积

采用 workqueue，多个消费者监听同一队列。
接收到消息以后，通过线程池，异步消费。

九、RabbitMQ 如何避免消息丢失

消费者的 ACK 机制。可以防止消费者丢失消息。
将消息进行持久化。但是要将消息持久化，前提是：队列、交换机都持久化。

生产者消息确认（Publisher Confirms）：

（一）原理

当生产者将消息发送到 RabbitMQ 服务器时，RabbitMQ 会给生产者一个确认回复，告知消息是否成功发送到 Broker。这样可以确保消息已经被正确接收和存储，防止消息在发送过程中丢失。

（二）示例代码

public class PublisherConfirmsExample {
    private final static String QUEUE_NAME = "confirm_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 开启消息确认模式
        channel.confirmSelect();

        String message = "This is a confirm message";
        // 发送消息
        channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));

        // 等待确认
        if (channel.waitForConfirms()) {
            System.out.println("Message sent successfully");
        } else {
            System.err.println("Message failed to send");
        }

        channel.close();
        connection.close();
    }
}

代码解释：

channel.confirmSelect();：开启消息确认模式，启用生产者确认机制。
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));：发送消息。
channel.waitForConfirms()：等待 RabbitMQ 的确认回复，如果收到确认回复，说明消息发送成功；如果未收到，则认为消息发送失败。

消息持久化：

（一）原理

将消息、队列和交换机进行持久化，即使 RabbitMQ 服务器重启或出现故障，也能保证消息不会丢失。

（二）实现步骤

持久化队列：在声明队列时，将 durable 参数设置为 true。

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

持久化交换机：在声明交换机时，将 durable 参数设置为 true。

channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);

持久化消息：在发送消息时，使用 MessageProperties.PERSISTENT_TEXT_PLAIN 等持久化属性。

channel.basicPublish(EXCHANGE_NAME, "routing_key", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));

示例代码：

public class PersistentMessageExample {
    private final static String QUEUE_NAME = "persistent_queue";
    private final static String EXCHANGE_NAME = "persistent_exchange";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 声明持久化的交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
        // 声明持久化的队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routing_key");

        String message = "This is a persistent message";
        // 发送持久化消息
        channel.basicPublish(EXCHANGE_NAME, "routing_key", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

代码解释：

channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);：声明持久化的交换机，true 表示该交换机在 RabbitMQ 重启后仍然存在。
channel.queueDeclare(QUEUE_NAME, true, false, false, null);：声明持久化的队列，true 表示该队列在 RabbitMQ 重启后仍然存在。
channel.basicPublish(EXCHANGE_NAME, "routing_key", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));：发送持久化消息，使用 MessageProperties.PERSISTENT_TEXT_PLAIN 保证消息的持久化。

消费者消息确认（ACK）：

（一）原理

消费者在处理完消息后，向 RabbitMQ 发送确认消息（ACK），告诉 RabbitMQ 消息已经被正确处理，RabbitMQ 会将该消息从队列中删除。如果消费者在处理消息时发生异常或宕机，且未发送 ACK，RabbitMQ 会将消息重新分发给其他消费者或在该消费者恢复后重新发送。

（二）实现步骤

对于手动 ACK 模式，在消费者处理完消息后调用 channel.basicAck 方法发送确认消息。

public class ConsumerAckExample {
    private final static String QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String receivedMessage = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + receivedMessage + "'");
            try {
                // 消息处理逻辑
                // 例如：存储消息、执行计算等
                System.out.println("Message processed successfully");
                // 手动确认消息
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            } catch (Exception e) {
                System.err.println("Message processing failed: " + e.getMessage());
                // 可以根据情况使用 basicNack 重新入队或 basicReject 拒绝消息
                // channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
            }
        };
        // 开启手动确认模式
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });
    }
}

代码解释：

channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });：开启手动确认模式，第二个参数 false 表示不自动确认消息。
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);：在消息处理成功后手动确认消息，确保消息被正确处理。
在 catch 块中可以根据具体情况使用 channel.basicNack 或 channel.basicReject 处理异常消息，如将消息重新入队或拒绝消息。

通过以上三种形式，可以从生产者、消息存储以及消费者三个方面来避免 RabbitMQ 中的消息丢失，提高消息队列系统的可靠性。同时，可以根据实际情况组合使用这些机制，以达到更好的效果。

十、生产者消费者执行流程

RabbitMQ 执行流程和生产者消费者执行流程两者的关系

包含关系：
- RabbitMQ 的执行流程包含了生产者消费者执行流程，同时涵盖了更多的内容，如 Exchange 的消息路由规则、集群和高可用等方面。
- 生产者消费者执行流程主要聚焦于生产者和消费者如何使用 RabbitMQ 进行消息的发送和接收，是 RabbitMQ 执行流程在应用程序端的体现。

示例代码：

public class RabbitMQExample {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        // 创建连接
        Connection connection = factory.newConnection();
        // 创建通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 生产者发送消息
        String message = "Hello, RabbitMQ!";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + message + "'");

        // 消费者接收消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String receivedMessage = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + receivedMessage + "'");
            // 手动确认消息
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); 
        };
        // 开启消息确认机制，设置为手动确认
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });
    }
}

代码解释：

首先使用 ConnectionFactory 创建一个连接工厂，并将 RabbitMQ 服务器的主机地址设置为 localhost。
通过该工厂创建连接并建立通道。
调用 channel.queueDeclare 声明一个队列，这里使用了默认或简单的设置，包括队列名称、是否持久化、是否独占、是否自动删除和额外参数。
生产者使用 channel.basicPublish 发送消息，将消息发送到默认的交换机并指定队列名称，同时设置消息内容的字节数组。
对于消费者，使用 DeliverCallback 来处理接收到的消息，将其字节数组转换为字符串并打印，随后通过 channel.basicAck 进行手动确认，确保消息处理完成后才从队列中移除。
最后使用 channel.basicConsume 开启消息消费，设置为手动确认，同时指定消息接收回调函数和取消消费的回调函数（这里未使用）。

十一、RabbitMQ 的高可用配置

（一）镜像队列

可以通过设置镜像队列来提高可用性，确保队列中的消息在多个节点上有副本。在 RabbitMQ 的管理界面或通过配置文件，可以将队列设置为镜像模式。

（二）集群模式

将多个 RabbitMQ 服务器组成一个集群，使用 rabbitmqctl 命令行工具或通过配置文件来配置集群。

rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

命令解释：

首先停止 RabbitMQ 应用程序，然后将当前节点加入到 rabbit@node1 节点的集群中，最后重新启动应用程序。

十二、RabbitMQ 的应用场景

异步任务处理：例如用户注册时发送欢迎邮件，将邮件发送任务作为消息放入 RabbitMQ，由专门的消费者来处理邮件发送，提高系统的响应速度和性能。在这种场景下，注册服务作为生产者将邮件发送任务封装成消息发送到 RabbitMQ 的队列中，而邮件发送服务作为消费者从队列中取出消息并完成邮件发送操作，从而将邮件发送操作与注册操作解耦，避免注册过程因邮件发送的延迟而阻塞。
日志收集和处理：应用程序将日志消息发送到 RabbitMQ，日志处理系统从队列中取出日志进行存储、分析和监控。不同的应用程序可以将日志消息发送到同一队列，由专门的日志处理服务来统一处理，实现日志的集中管理和处理。
系统解耦：不同的系统之间通过 RabbitMQ 传递消息，实现系统之间的解耦，提高系统的可维护性和可扩展性。例如订单系统和库存系统，当有订单生成时，订单系统作为生产者将订单消息发送到 RabbitMQ，库存系统作为消费者从队列中获取订单消息并进行库存调整操作，而不是直接调用库存系统的接口，这样当订单系统或库存系统需要升级或维护时，彼此之间的影响较小。

十三、依赖和配置

（一）Maven 依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.13.1</version>
</dependency>

代码解释：

在项目的 pom.xml 文件中添加上述依赖，这样可以使用 RabbitMQ 的 Java 客户端库。同时，确保 RabbitMQ 服务器已经启动，可以在 localhost 上访问，也可以根据实际情况修改 ConnectionFactory 中的 setHost 参数。

原文地址：https://blog.csdn.net/xxiaobaibaibai/article/details/145144011

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

上一篇：Pytest入门—allure生成报告
下一篇：【0391】Postgres内核 checkpointer process ① 启动初始化

基于Python+django的社区爱心养老管理系统设计与实现
完整源码项目包获取→点击文章末尾名片！
阅读更多2025-01-16
STM32 物联网智能家居 (二)-开发环境及工程搭建(STM32CubeMX)
这篇文章我们具体会讲到整个项目需要哪些硬件模块、软件环境、搭建最小工程时用到STM32CubeMX的详细配置指导，以及最后用Keil 5工具进行编译工程的教学，- 中控屏启动后，自动连接家里的路由器，
阅读更多2025-01-16
论文略读：ASurvey of Large Language Models for Graphs
2024 KDD。
阅读更多2025-01-16
java 设计模式工厂模式
什么是工厂模式，为什么使用工厂模式，工厂模式作用是什么，springboot那里地方使用了工厂模式
阅读更多2025-01-16
硬件学习笔记--29 IEC62052-11 相关内容介绍
IEC62052-11是一项由国际电工委员会（IEC）制定的国际标准，主要涉及电能计量装置的测量和限制技术要求。该标准详细规定了电能计量装置的各个方面，包括测量原理、计算方法、测量误差、输出信号等。
阅读更多2025-01-16
C# (图文教学)在C#的编译工具Visual Studio中使用SQLServer并对数据库中的表进行简单的增删改查--14
在C#的编译工具Visual Studio中使用SQLServer在VS中对SQLServer中的Students表进行简单的增删改查
阅读更多2025-01-16
Go Ebiten小游戏开发：贪吃蛇
贪吃蛇的核心玩法是控制一条蛇在网格中移动，吃掉随机生成的食物，每吃一个食物蛇身会变长，同时得分增加。如果蛇撞到墙壁或自己的身体，游戏结束。本项目使用 Go 语言和 Ebiten 游戏引擎实现。Ebit
阅读更多2025-01-16
【Python】第一弹---解锁编程新世界：深入理解计算机基础与Python入门指南
计算机基础概念：什么是计算机，什么是编程, 编程语言有哪些；Python 背景知识：Python 是咋来的， Python 都能干啥， Python 的优缺点，Python 的前景(钱景)咋样；
阅读更多2025-01-16
Zookeeper 数据迁移实战：基础环境搭建与高效迁移方案全览
Zookeeper 数据迁移实战：基础环境搭建与高效迁移方案全览
阅读更多2025-01-16
【机器学习】数据拟合-最小二乘法（Least Squares Method）
最小二乘法是一种广泛使用的数据拟合方法，用于在统计学和数学中找到最佳拟合曲线或模型，使得观测数据点与模型预测值之间的误差平方和最小化。
阅读更多2025-01-16

RabbitMQ原理、使用与实践指南

一、概念

二、特点

三、AMQP 和 JMS MQ 的区别和联系

四、RabbitMQ 的核心组件

五、RabbitMQ 的执行流程

（一）生产者发送消息流程

（二）消费者接收消息流程

六、RabbitMQ 的六种消息模式

七、RabbitMQ 的消息确认机制（ACK）

（一）ACK 机制类型

自动确认（Auto ACK）

手动确认（Manual ACK）

否定确认（Nack）

拒绝消息（Reject）

（二）ACK 机制选择

（三）注意事项

八、RabbitMQ 如何避免消息堆积

九、RabbitMQ 如何避免消息丢失

生产者消息确认（Publisher Confirms）：

（一）原理

（二）示例代码

代码解释：

消息持久化：

（一）原理

（二）实现步骤

示例代码：

代码解释：

消费者消息确认（ACK）：

（一）原理

（二）实现步骤

代码解释：

十、生产者消费者执行流程

RabbitMQ 执行流程 和 生产者消费者执行流程 两者的关系

示例代码：

十一、RabbitMQ 的高可用配置

（一）镜像队列

（二）集群模式

十二、RabbitMQ 的应用场景

十三、依赖和配置

（一）Maven 依赖

相关文章

RabbitMQ 执行流程和生产者消费者执行流程两者的关系