215 lines
10 KiB
Markdown
215 lines
10 KiB
Markdown
---
|
||
tags: [MQ, RabbitMQ, 消息队列, AMQP]
|
||
create time: 2026-05-24 19:52
|
||
---
|
||
|
||
# RabbitMQ
|
||
|
||
## 概述
|
||
|
||
RabbitMQ 是基于 Erlang/OTP 实现的开源消息代理，实现了 AMQP 0-9-1 协议。它以灵活的路由能力、丰富的协议支持和成熟的插件生态著称，是企业级消息中间件的经典选择。
|
||
|
||
## 正文
|
||
|
||
### 1. 架构与 AMQP 协议
|
||
|
||
RabbitMQ 使用 Erlang 语言编写，天然具备高并发和软实时特性。其核心概念围绕 AMQP 0-9-1 协议展开：
|
||
|
||
- **Connection**：TCP 长连接，客户端与 Broker 之间的一条物理连接，包含认证和 TLS 握手。
|
||
- **Channel**：在 Connection 上的多路复用虚拟连接。多个 Channel 共享同一个 TCP 连接，避免频繁创建/销毁 TCP 的开销。一个线程对应一个 Channel。
|
||
- **Virtual Host（vhost）**：逻辑隔离单元，类似数据库中的 schema，不同 vhost 的 Exchange 和 Queue 完全隔离。
|
||
- **Exchange**：接收 Producer 发来的消息，根据路由规则分发到 Queue。
|
||
- **Queue**：消息的最终存储位置，Consumer 从 Queue 拉取或被推送消息。
|
||
- **Binding**：连接 Exchange 和 Queue 的规则，定义路由条件。
|
||
|
||
```mermaid
|
||
graph LR
|
||
Producer["Producer"] -->|"publish"| Exchange["Exchange"]
|
||
Exchange -->|"binding rule"| Q1["Queue A"]
|
||
Exchange -->|"binding rule"| Q2["Queue B"]
|
||
Q1 -->|"consume"| C1["Consumer A"]
|
||
Q2 -->|"consume"| C2["Consumer B"]
|
||
|
||
style Exchange fill:#4A90D9,color:#fff
|
||
```
|
||
|
||
### 2. 四种 Exchange 类型
|
||
|
||
Exchange 的类型决定了消息如何路由到 Queue，这是 RabbitMQ 最灵活的设计：
|
||
|
||
**Direct Exchange**：精确匹配。消息的 Routing Key 与 Binding Key 完全一致时才路由。适合点对点定向投递。
|
||
|
||
**Fanout Exchange**：广播模式。忽略 Routing Key，将消息投递到所有绑定的 Queue。适合事件通知、广播场景。
|
||
|
||
**Topic Exchange**：通配符匹配。Routing Key 和 Binding Key 支持 `*`（匹配一个词）和 `#`（匹配零或多个词）。适合按主题分类订阅，如 `order.*.created`。
|
||
|
||
**Headers Exchange**：基于消息 Header 属性匹配（而非 Routing Key）。支持 `x-match=all`（所有条件匹配）和 `x-match=any`（任一条件匹配）。灵活但性能不如前三种。
|
||
|
||
```mermaid
|
||
graph TB
|
||
P["Producer"] --> EX_D["Direct Exchange<br/>routing_key = order"]
|
||
P --> EX_F["Fanout Exchange<br/>广播"]
|
||
P --> EX_T["Topic Exchange<br/>order.*.created"]
|
||
|
||
EX_D -->|"rk = order"| QA["Queue A"]
|
||
EX_F --> QA
|
||
EX_F --> QB["Queue B"]
|
||
EX_F --> QC["Queue C"]
|
||
EX_T -->|"order.pay.created"| QB
|
||
EX_T -->|"order.ship.created"| QD["Queue D"]
|
||
|
||
style EX_D fill:#4A90D9,color:#fff
|
||
style EX_F fill:#7B68EE,color:#fff
|
||
style EX_T fill:#F5A623,color:#fff
|
||
```
|
||
|
||
> [!question] 生产环境中，什么时候该用 Topic Exchange 而不是 Direct Exchange？
|
||
> 当消费者需要按模式订阅一类消息时用 Topic。比如日志系统中，`log.error.*` 可以匹配所有 error 级别的日志，而 Direct 只能精确匹配一个 routing key。如果你的路由规则是固定的、一一对应的，Direct 更简单高效。
|
||
|
||
### 3. 高级特性
|
||
|
||
**TTL（Time-To-Live）**：
|
||
- **消息 TTL**：通过 `x-message-ttl` 设置队列级别 TTL，或在发布时通过 `expiration` 属性设置单条消息 TTL。过期消息被丢弃或进入死信队列。
|
||
- **队列 TTL**：通过 `x-expires` 设置，队列在空闲（无消费者、无声明）超过指定时间后自动删除。适合临时队列。
|
||
|
||
**死信队列（DLX, Dead Letter Exchange）**：消息在以下情况会成为"死信"：被消费者拒绝（reject/nack 且 requeue=false）、消息 TTL 到期、队列达到最大长度。死信会被路由到配置的 DLX 对应的队列，实现延迟重试、异常消息归档等模式。
|
||
|
||
**延迟队列**：RabbitMQ 原生不支持延迟消息（不像 RocketMQ 有延迟级别）。社区提供了 `rabbitmq_delayed_message_exchange` 插件，消息在 Exchange 中暂存，到期后再投递到目标队列。常用于订单超时取消、定时提醒等场景。
|
||
|
||
> [!question] 死信队列和延迟队列有什么关系？
|
||
> 延迟队列的一种经典实现方式就是"利用消息 TTL + 死信队列"：将消息发到一个没有消费者的队列并设置 TTL，到期后消息变成死信，被路由到 DLX 绑定的真正消费队列。但这有个缺点——队列头部消息未过期会阻塞后面的消息（因为 RabbitMQ 只检查队头）。延迟消息插件则用定时器解决这个问题。
|
||
|
||
### 4. 集群模式
|
||
|
||
RabbitMQ 支持多种集群模式，可靠性逐步递增：
|
||
|
||
**普通集群**：所有节点共享元数据（Exchange、Binding 等），但 Queue 的数据只存在于声明它的那个节点。其他节点收到消息后需要跨节点转发，存在单点风险。
|
||
|
||
**镜像队列（Mirrored Queue）**：在普通集群基础上，将 Queue 数据同步到多个节点。一个 Master + 若干 Slave，所有读写都经过 Master。缺点是：所有操作都由 Master 串行处理，性能受限于 Master 节点；同步方式是"发一份拷贝"，网络开销大；Slave 只是热备，不承担读流量。**这就是镜像队列性能差的根本原因——单 Master 瓶颈。**
|
||
|
||
**Quorum Queue**：基于 Raft 共识协议的队列类型，是镜像队列的现代替代方案。Leader 负责接收写入，消息被复制到多数派 Follower 后才确认。Follower 可以分担读流量（`x-queue-leader-locator=balanced`），Leader 故障后自动选举新 Leader。**Quorum Queue 的改进在于：Raft 共识比简单的主从复制更可靠，Follower 可以参与读操作，且日志复制有明确的多数派确认语义。**
|
||
|
||
> [!question] 思考题：RabbitMQ 的镜像队列为什么性能不好？Quorum Queue 是如何改进的？
|
||
>
|
||
> **提示**：镜像队列的核心问题是"所有流量都走 Master"——写入要 Master 确认，读取也要 Master 响应，Slave 只是被动同步。当 Master 成为瓶颈时，加 Slave 不能提升吞吐。Quorum Queue 用 Raft 日志复制替代简单的消息拷贝，写入由多数派确认（而非 Master 独占），且支持从 Follower 读取，分散了压力。
|
||
|
||
### 5. RabbitMQ Stream
|
||
|
||
Stream 是 RabbitMQ 3.9 引入的全新队列类型，对标 Kafka 的流式存储模型：
|
||
|
||
- 消息以日志形式持久化，支持多次回放（不像普通队列消费即删除）。
|
||
- 使用 offset 而非 ACK 来跟踪消费进度，支持从任意位置开始消费。
|
||
- 性能远超传统队列，适合高吞吐的日志/事件流场景。
|
||
- 通过 AMQP 1.0 或专用 Stream 协议访问。
|
||
|
||
Stream 本质上让 RabbitMQ 同时具备了"传统消息队列"和"流处理平台"两种能力。
|
||
|
||
### 6. 插件生态
|
||
|
||
RabbitMQ 的可扩展性通过插件机制实现：
|
||
|
||
| 插件 | 作用 |
|
||
|------|------|
|
||
| **Management UI** | Web 管理界面，监控队列、Exchange、连接，管理策略和权限 |
|
||
| **Prometheus 插件** | 暴露 Prometheus 格式的监控指标，配合 Grafana 看板 |
|
||
| **Shovel** | 单向消息搬运，将消息从一个 Broker 转发到另一个，适合跨机房同步 |
|
||
| **Federation** | 跨集群消息联邦，支持 Exchange/Queue 级别的联邦，比 Shovel 更灵活，支持按需连接 |
|
||
|
||
### 7. 消息确认机制
|
||
|
||
RabbitMQ 提供了完整的可靠投递保证：
|
||
|
||
**Publisher Confirm（发布确认）**：Producer 将 Channel 设置为 confirm 模式后，Broker 成功将消息写入所有镜像（或 Quorum 多数派）后返回确认。支持同步 confirm 和异步 confirm（批量/单条）。注意区分 confirm 和事务——confirm 更轻量，性能更好。
|
||
|
||
**Consumer ACK（消费确认）**：消费者处理完消息后显式调用 `basicAck`，Broker 才从队列中删除消息。如果消费者崩溃未 ACK，消息会被重新投递。支持 `basicNack` / `basicReject` 拒绝消息并决定是否 requeue。
|
||
|
||
**Return 机制**：当消息无法路由到任何 Queue（没有匹配的 Binding），且 `mandatory=true` 时，Broker 通过 Return 回调通知 Producer。配合 `alternate-exchange` 可以将无法路由的消息转入备用 Exchange。
|
||
|
||
```go
|
||
// 使用 amqp091-go 的完整发布确认 + 手动 ACK 示例
|
||
package main
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
amqp "github.com/rabbitmq/amqp091-go"
|
||
"log"
|
||
"time"
|
||
)
|
||
|
||
func main() {
|
||
// 建立连接
|
||
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
|
||
if err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
defer conn.Close()
|
||
|
||
ch, err := conn.Channel()
|
||
if err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
defer ch.Close()
|
||
|
||
// 声明 Quorum Queue（生产环境推荐）
|
||
_, err = ch.QueueDeclare("order-queue", true, false, false, false, amqp.Table{
|
||
"x-queue-type": "quorum",
|
||
})
|
||
if err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
|
||
// ===== 发布确认 =====
|
||
// 将 Channel 设置为 confirm 模式
|
||
if err := ch.Confirm(false); err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
// 获取确认通知 channel
|
||
confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))
|
||
|
||
// 发布消息
|
||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
|
||
err = ch.PublishWithContext(ctx, "", "order-queue", false, false, amqp.Publishing{
|
||
ContentType: "application/json",
|
||
Body: []byte(`{"order_id": "1001", "amount": 99.9}`),
|
||
DeliveryMode: amqp.Persistent, // 持久化消息
|
||
})
|
||
if err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
|
||
// 等待 Broker 确认
|
||
conf := <-confirms
|
||
if !conf.Ack {
|
||
log.Fatal("message was nacked by broker")
|
||
}
|
||
fmt.Println("publish confirmed")
|
||
|
||
// ===== 消费端手动 ACK =====
|
||
msgs, err := ch.Consume("order-queue", "", false, false, false, false, nil)
|
||
if err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
|
||
for msg := range msgs {
|
||
fmt.Printf("received: %s\n", msg.Body)
|
||
// 处理完成后手动确认，false 表示只确认当前消息
|
||
if err := msg.Ack(false); err != nil {
|
||
log.Printf("ack failed: %v", err)
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
## 关联笔记
|
||
|
||
- [[03-协议与标准/5-AMQP-协议|AMQP 协议]]
|
||
- [[04-存储引擎/11-RabbitMQ-消息存储|RabbitMQ 消息存储]]
|
||
- [[05-可靠性保障/12-MQ-消息确认与持久化|MQ 消息确认与持久化]]
|
||
- [[05-可靠性保障/16-MQ-死信队列与消息回溯|MQ 死信队列与消息回溯]]
|
||
- [[06-高级特性/17-MQ-延迟消息与定时消息|MQ 延迟消息与定时消息]]
|
||
- [[12-架构与实战/44-MQ-高可用架构|MQ 高可用架构]]
|
||
- [[07-主流MQ对比/27-MQ-选型对比|MQ 选型对比]]
|