---
url: /rabbitmq.md
---
[RabbitMQ-config](./RabbitMQ-config.md)

## 工作原理

## exchange

direct exchange

fanout exchange

topic exchange

header exchange

ConsistentHash exchange  专注于‌流量权重控制‌与‌一致性哈希算法‌的结合

## 工作模式

simple

work queues

publish/subscribe

topics

## 消息持久化需做到三点

Exchange 设置持久化

Queue 设置持久化

Message 持久化发送：发送消息设置发送模式 deliveryMode=2，代表持久化消息

## amqp

AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
AMQP协议更多用在企业系统内，对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。

## 一、协议定义与目标

AMQP（Advanced Message Queuing Protocol，高级消息队列协议）是一个**开放标准的应用层协议**，旨在为消息中间件提供统一、标准化的通信范式。其根本目标是实现**跨平台、跨语言的互操作性**，降低分布式系统中应用集成的复杂度，支撑可靠、异步的消息传递 \[ref_1]。

## 二、核心架构组件

| 组件             | 职责说明                                                                 | 典型实现/约束示例                          |
|------------------|--------------------------------------------------------------------------|-------------------------------------------|
| **消息代理（Broker）** | 协议中枢：负责连接管理、消息路由、持久化存储、流控与监控                     | RabbitMQ、Qpid、Apache ActiveMQ（部分兼容） |
| **生产者（Producer）** | 消息源头：按AMQP语义构造并发布消息至Exchange，无需感知下游拓扑                   | Java客户端通过`Channel.basicPublish()`调用    |
| **消费者（Consumer）** | 消息终端：订阅Queue并处理消息，支持手动ACK/NACK以控制重投逻辑                  | Spring AMQP中`@RabbitListener`注解驱动消费   |
| **交换机（Exchange）** | 路由引擎：依据Type（direct/topic/fanout/headers）及Binding Key决策消息分发路径     | RabbitMQ内置4类Exchange，不可自定义类型         |
| **队列（Queue）**      | 有界缓冲区：消息暂存单元，支持Durable（磁盘持久化）、Exclusive（私有）、Auto-delete等属性 | 声明时需显式设置`durable=true`保障重启不丢失     |
| **虚拟主机（Virtual Host）** | 逻辑隔离域：提供独立的Exchange/Queue命名空间与ACL权限体系，实现多租户资源管控       | RabbitMQ中以`/`或`/test`形式标识，权限粒度达configure/read/write |

## 三、标准工作流程

遵循严格四阶段链路：

1. **生产者 → Exchange**：调用`basic.publish`发送消息，携带`routingKey`与`headers`；
2. **Exchange → Queue**：根据绑定规则（Binding）匹配`routingKey`，将消息路由至一个或多个Queue；
3. **Queue → Consumer**：消息入队后等待拉取（`basic.get`）或推送（`basic.consume`）；
4. **Consumer确认**：处理完成后发送`basic.ack`（成功）或`basic.nack`（失败重试/丢弃），否则Broker保留消息直至超时。

## 四、关键特性与企业级能力

* **异步解耦**：生产者无需等待消费者响应，提升系统吞吐与容错性；
* **可靠性保障**：支持消息持久化（Queue+Message双重标记）、发布确认（Publisher Confirms）、消费者手动应答；
* **事务支持**：通过`tx.select`/`tx.commit`实现跨信道原子操作（性能代价高，生产环境慎用）；
* **安全机制**：集成TLS 1.2+加密传输与SASL认证（PLAIN/EXTERNAL），满足PCI-DSS等合规要求；
* **可扩展架构**：支持RabbitMQ镜像队列（Mirrored Queues）与Federation/Shovel插件，实现跨机房高可用与水平伸缩。

## 五、典型应用场景

* **金融领域**：证券行情实时分发（要求亚秒级端到端延迟 + 精确一次语义）；
* **物联网（IoT）**：车联网TSP平台日均处理亿级车辆心跳/诊断上报；
* **微服务架构**：Spring Cloud生态中基于Spring AMQP构建事件驱动架构（EDA），解耦订单、库存、物流服务；
* **企业集成**：跨异构系统（Java/.NET/Python）实现松耦合数据同步与事件通知。

## 问题方案

1、 防止消息重复消费：
消息幂等性：确保消费者的处理逻辑具备幂等性，即多次处理相同的消息不会产生额外的影响。这样即使消息被重复消费，也不会导致不一致状态。
消息去重：使用消息去重机制来检查已经处理过的消息，避免重复处理。例如，维护一个消息ID的集合，在消费者端检查接收到的消息ID是否已存在于集合中。
消息确认机制：确保消息在被消费者成功处理后才被标记为已消费。这可以减少消息的重复传递。RabbitMQ提供了消费者应答（ack）机制，只有在消息被成功处理后才发送ack，否则RabbitMQ会重新将消息放入队列。

2、防止消息丢失：
持久化消息：将消息的delivery_mode设置为2，表示将消息持久化到磁盘上。这样即使RabbitMQ服务器重启，消息也不会丢失。
消息确认机制：在消费者端开启消息确认机制，确保消息被正确处理后才发送ack。如果消费者在处理消息时崩溃，RabbitMQ会将未确认的消息重新放回队列，等待其他消费者处理。
发送消息时开启事务：确保消息成功发送到RabbitMQ服务器。如果事务提交成功，消息就会被发送到消息队列中；如果事务回滚，消息就会被丢弃。
设置备份交换机：当消息无法被路由到目标队列时，可以将其发送到备份交换机中，以防止消息丢失。

3、确保消息的顺序性：
使用单个消费者：让单个消费者处理队列中的所有消息，这样可以确保消息按照它们进入队列的顺序被处理。但这种方法在处理大量消息时可能会成为性能瓶颈。
使用锁机制：防止多个消费者同时访问同一消息，但可能增加系统的复杂性和开销。
结合使用事务：在事务期间，消费者会锁定消息，进行一些处理，然后提交事务。如果事务失败，消息将被回滚到队列中

开箱即用；优先级队列；延迟队列；死信队列；消息重试；消息回溯；消息堆积 + 持久化；消息跟踪；消息过滤；消息顺序性；安全机制；消息幂等性；事务性消息等。

## 基本概念

Broker： 简单来说就是消息队列服务器实体

Exchange： 消息交换机，它指定消息按什么规则，路由到哪个队列

Queue： 消息队列载体，每个消息都会被投入到一个或多个队列

Binding： 绑定，它的作用就是把exchange和queue按照路由规则绑定起来

Routing Key： 路由关键字，exchange根据这个关键字进行消息投递

VHost： vhost 可以理解为虚拟 broker ，即 mini-RabbitMQ server。其内部均含有独立的 queue、exchange 和 binding 等，

但最最重要的是，其拥有独立的权限系统，可以做到 vhost 范围的用户控制。当然，从 RabbitMQ 的全局角度，vhost 可以作为不同权限隔

离的手段（一个典型的例子就是不同的应用可以跑在不同的 vhost 中）。

Producer： 消息生产者，就是投递消息的程序

Consumer： 消息消费者，就是接受消息的程序

Channel： 消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务

由Exchange、Queue、RoutingKey三个才能决定一个从Exchange到Queue的唯一的线路。

## 最佳实践

生产端开启异步Confirm+Return机制，对NACK/Return消息做重试、

落库兜底 (return 机制:当 Mandatory = true 时，消息路由失败时，Broker 会将消息返回给生产者；)
核心业务必须开启Exchange、Queue、消息三级持久化

消费端必须使用手动ACK模式，配合QoS预取机制

业务异常时，禁止无限重试入队（避免消息循环），应设置重试次数，超过次数后转为死信

## 消息可靠传输

同时满足以下条件才能保证

队列设置 durable=true

交换机设置 durable=true

消息 deliveryMode=2

消费者使用手动确认（manual ACK）‌

## 镜像队列（Mirrored Queue）

提供消息队列的高可用性。镜像 队列通过在多个节点上复制消息队列的数据来实现

## Lazy Queues（惰性队列）

RabbitMQ 3.6.0 版本开始，引入了 Lazy Queues（惰性队列）

RabbitMQ 3.12 版本之后，LazyQueue 已经成为所有队列的默认格式。这种模式可以有效地解决消息积压问题，提高 RabbitMQ 的性能和稳定性

## MQ消息丢失、重复、积压

消息丢失通过三段机制解决：生产者开启重试与return确认机制confirm、MQ开启持久化存储durable、消费者手动ACK确认消费完成。

消息重复由网络重试导致，核心解决方案为业务幂等，通过唯一消息ID、状态机、数据库唯一索引实现幂等去重。

消息积压由消费慢、代码阻塞、异常报错导致，解决方案为优化消费逻辑、扩容消费者节点、扩容分区、异常消息转入死信队列兜底