10 KiB
tags, create time
| tags | 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 的规则，定义路由条件。
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（任一条件匹配）。灵活但性能不如前三种。
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。
// 使用 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)
}
}
}