ozine's book library

主题

消息队列其他选型：从路由、重放和业务交接来选择 ​

“哪个消息队列最好”很难给出有用答案。更具体的问法是：消息要按规则分发给多个服务，还是要长期保留供重放？消费者离线后必须补收吗？最棘手的问题是路由、订单提交与发布之间的裂缝，还是大量租户的存储和隔离？

本文逐一解释 RabbitMQ、RocketMQ、Core NATS、JetStream、Pulsar 和 Redis Streams 的使用入口及边界。版本基线是 RabbitMQ 4.2、RocketMQ 5.x、Pulsar 4.0.x；NATS 与 Redis 使用于 2026-10-13 核对的官方滚动文档，不把页面中的新特性默认归给旧服务器。没有运行这些服务器，文中的 Go 小程序只验证自建路由与待确认记录模型。

先用同一套问题比较 ​

选择常见协议/API 与组织方式消费进度和保存方式值得优先评估的需求随之而来的负担
RabbitMQAMQP 0-9-1：exchange → binding → queue；另有其他协议与 Stream普通队列用逐条/批量 ack；复制与持久化依队列类型复杂路由、工作分发、已有 AMQP 生态拓扑、confirm、ack、重投与队列资源治理
RocketMQTopic、消息类型、消费组；不同代际 SDK/API按消费模型跟踪进度、重试与确认事务消息、延迟消息、按业务组有序处理本地事务回查、消息类型配置、SDK/服务端能力匹配
Core NATSSubject 发布订阅、queue group、request-reply核心实时转发，不提供离线重放存储允许丢失的在线通知、低开销服务通信离线、慢消费者、重连期间的业务策略
JetStreamNATS 上增加 stream 与有状态 consumer可配置存储/复制/保留、发布确认与消费 ack已用 NATS，又需要持久消息和重放需要管理存储、consumer、ack 时限和重投
PulsarTopic、subscription、Message ID；专用客户端协议持久 Topic 数据和游标由存储层支撑多租户、多订阅模型、计算与存储分离Broker、BookKeeper、元数据服务等多层运维
Redis StreamsRESP：XADD / XREADGROUP / XACK 等Stream 条目、消费组与待确认列表 PEL已有 Redis，需求规模与可靠性边界都可控自行组织接管、重试、隔离、容量与持久化策略

这张表用于缩小候选，不是性能排名。“轻量”“云原生”也不是可靠性等级。一个可执行文件容易启动，不代表多副本持久化、跨故障域恢复、权限和备份同样无需设计。

RabbitMQ：先画出交换机和队列的关系 ​

AMQP 0-9-1 的生产者发布到 exchange，exchange 根据类型和 binding 决定路由到哪些 queue。消费者从 queue 取消息。队列负责保存等待处理的内容，交换机负责匹配路径。[1]

Exchange 类型匹配什么例子和用途
directrouting key 精确相等invoice.ready 发给绑定相同键的账单队列
fanout忽略 routing key，给所有绑定目标通知与审计各自得到一份
topic按 . 分段的模式；* 一段，# 零段或多段order.* 匹配 order.paid；order.# 也匹配 order.paid.eu
headers根据消息 header 与匹配条件路由维度适合放在属性里，而非拼接字符串

默认交换机名字是空字符串；声明队列时会自动建立“队列名作为 routing key”的绑定。因此看起来像直接发给某个队列，实际上仍经过交换机。交换机和队列的 durable 属性也不同：持久交换机保存的是拓扑，不能替消息本身提供持久化。[1:1]

正在绘制图示…

可切换适合窗口或原始尺寸
查看图源代码
flowchart LR
    P["订单服务<br/>routing key: order.paid"] --> X["events<br/>topic exchange"]
    X -- "binding: order.paid" --> Q1["billing 队列"]
    X -- "binding: order.#" --> Q2["audit 队列"]
    Q1 --> C1["账单实例 A"]
    Q1 --> C2["账单实例 B"]
    Q2 --> C3["审计服务"]
放大图示

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码
flowchart LR
    P["订单服务<br/>routing key: order.paid"] --> X["events<br/>topic exchange"]
    X -- "binding: order.paid" --> Q1["billing 队列"]
    X -- "binding: order.#" --> Q2["audit 队列"]
    Q1 --> C1["账单实例 A"]
    Q1 --> C2["账单实例 B"]
    Q2 --> C3["审计服务"]

A、B 竞争处理账单队列，审计服务独立处理另一队列。一个队列有两个消费者，不等于每个消费者各收到一份；要两套业务独立处理，就需要独立队列或相应独立订阅结构。

普通工作队列的常用操作是：声明 exchange/queue/binding，生产端开启 confirm 并处理无法路由的 return，消费端关闭自动确认、设置 prefetch，业务提交之后 ack。临时失败可负确认并按策略重试；一直立即重新入队会让毒消息形成热循环。[2]

队列类型影响取舍。RabbitMQ 4.x 的 classic queue 不是复制队列，旧式 classic mirroring 已移除；quorum queue 采用复制机制，适合需要数据安全的工作队列；Stream 则是可重放的日志形态。不能拿 classic queue 的成本谈 quorum queue 的保证，也不能因为普通 queue 确认后移除消息，就说 RabbitMQ 整个产品不能重放。[3]

优点是路由表达直接，工作分发和确认模型清楚。代价是 exchange、queue、binding、policy 增多后，需要把拓扑当作受版本管理的配置；多个消费者、优先级与重投还会影响业务完成顺序。

RocketMQ：事务消息解决生产交接，消费正确性仍要自己做 ​

RocketMQ 把普通、顺序、延迟和事务等消息行为放进明确的模型里。评估时先核对 Topic 的消息类型、部署版本，以及所选 Go SDK 属于哪个协议代际、支持哪些能力，不能只看服务端支持就认为客户端一定能调用。[4]

事务消息最适合解释“数据库提交后，消息没发出去”的裂缝。它的基本过程是先向 Broker 写入对消费者不可见的半消息，然后执行生产者本地事务，再告诉 Broker 提交或回滚这条消息。结果不明时，Broker 回查生产者，生产者从持久业务记录判断本地事务结果。[4:1]

正在绘制图示…

可切换适合窗口或原始尺寸
查看图源代码
sequenceDiagram
    participant P as 生产者
    participant B as RocketMQ Broker
    participant D as 本地数据库
    participant C as 消费者
    P->>B: 半消息（暂不可消费）
    B-->>P: 接收结果
    P->>D: 执行并提交本地事务
    D-->>P: 事务结果
    P--xB: 二阶段结果丢失
    B->>P: 回查事务结果
    P->>D: 按业务事务 ID 查询持久记录
    D-->>P: 已提交
    P->>B: Commit
    B->>C: 消息变为可消费
放大图示

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码
sequenceDiagram
    participant P as 生产者
    participant B as RocketMQ Broker
    participant D as 本地数据库
    participant C as 消费者
    P->>B: 半消息（暂不可消费）
    B-->>P: 接收结果
    P->>D: 执行并提交本地事务
    D-->>P: 事务结果
    P--xB: 二阶段结果丢失
    B->>P: 回查事务结果
    P->>D: 按业务事务 ID 查询持久记录
    D-->>P: 已提交
    P->>B: Commit
    B->>C: 消息变为可消费

回查必须在原进程重启后仍有依据，不能只检查内存里的布尔值。查不到记录也未必就等于“确定回滚”：本地事务可能还在进行，查询副本可能落后，或者查询本身失败。应按具体业务记录和超时规则返回明确状态，不能把暂时未知伪装成否定结果。

这里的 Rollback 作用于半消息的可见性；Broker 不会撤销已经发生的本地数据库提交，更不会撤销下游已经发出的邮件。事务消息协调的是生产者本地事务与消息发布之间的最终一致性，消费端仍要处理重试、幂等、永久失败和对账。回查和半消息都有时间与次数边界，所以不可把它理解为无限期等待之后必然自动一致。[4:2]

延迟消息的用途是“将来再检查”，例如到期检查订单是否仍未支付。5.x 文档按毫秒 Unix 时间戳设置可投递时间，并规定允许范围和异常时间的处理；4.x 常见延迟级别用法不能直接当成同一接口。时间戳单位、最大期限、Topic 类型都应在生产前校验；消费者执行时再检查业务状态，迟到或重复不应取消已支付订单。[5]

当团队确实需要这些消息行为时，RocketMQ 能减少自建调度和生产交接协议的工作。换来的责任是维护可靠的事务回查、排查长期未决消息，以及处理顺序消息中某个失败项阻塞后续的取舍。它不是数据库分布式事务的无条件替代品。

NATS 与 JetStream：同一生态里的两种承诺 ​

Core NATS 按 subject 实时发布订阅，queue group 让一个组内的订阅者分担消息，request-reply 在消息层构造请求与响应。它没有替离线消费者保留和重放消息的承诺，核心投递是 at-most-once。对于可以接受丢失的在线状态通知，这种取舍可能合适；订单事实如果需要离线补收，应进一步评估 JetStream。[6]

NATS subject 也用 . 分段，但通配符不能照搬 RabbitMQ：* 匹配一段，尾部 > 匹配一段或多段，RabbitMQ 的 # 则可匹配零段或多段。[7]

下面是我们自己编写的模式匹配器输出，不是两个服务器的调用结果：

go
fmt.Println("rabbit order.# / order:", rabbitMatch("order.#", "order"))
fmt.Println("nats order.> / order:", natsMatch("order.>", "order"))
fmt.Println("rabbit order.* / order.paid:", rabbitMatch("order.*", "order.paid"))
fmt.Println("nats order.* / order.paid.eu:", natsMatch("order.*", "order.paid.eu"))
text
rabbit order.# / order: true
nats order.> / order: false
rabbit order.* / order.paid: true
nats order.* / order.paid.eu: false

第三、四行说明多一个层级会改变匹配。设计 subject 时要把事件类型、租户、区域等维度放在约定的位置；中途改变层级会让订阅漏收，而程序未必报错。教学匹配器以各分段非空为输入前提；不验证空分段，也不实现完整协议校验或大规模路由索引。

JetStream 多出的不是一个开关，而是一组持久状态 ​

对象/动作作用要核对的设置
Stream捕获匹配 subject 的消息内存/文件存储、副本数、容量、保留策略
Consumer对 stream 的有状态消费视图durable 名称、起点、过滤、ack 策略
JetStream publish获得服务端发布确认检查确认/错误；使用稳定消息 ID 处理重复
Pull consumer按客户端能力取一批，或由 SDK 持续拉取批量、等待、在途上限
Ack / Nak / 处理中通知完成、重试或延长处理相关状态ack 时限、退避、最大交付次数

JetStream consumer 不等于一个 Go 进程；多个客户端可以共享同一个 consumer 分担工作，多个 consumer 又可以独立读取 stream。代码表面是回调，也可能是 SDK 在后台持续发 pull 请求。[8][9]

保留策略也会改变语义：按限制保留、按订阅兴趣保留、按工作确认移除，适合不同目的。把本来用于长期回放的流改成工作队列保留策略，消费成功后可能就没有同样的回放数据了。副本、文件存储、备份和磁盘容量要单独评估，不能因 Core NATS 启动简单就省略这些工作。[10]

JetStream 的消息 ID 去重有配置窗口，双向确认可以让客户端获知服务端已收到消费 ack。这些机制减少指定协议环节的重复，不会把数据库更新和 ack 合成原子事务。达到最大交付次数也不等于自动完成了一套“安全搬到 DLQ”的业务流程，需要处理 advisory、隔离和人工恢复。[11][12]

Pulsar：订阅方式丰富，存储层也需要专门运维 ​

Pulsar 的持久 Topic 由 Broker 服务读写请求，数据由 BookKeeper bookie 节点保存，元数据服务管理所有权与协调信息。把服务和持久存储拆开，让它们可以按不同需求扩展；代价是生产环境要维护更多组件和更清楚的故障链。[13]

一个 Topic 可以有多个 subscription，每个订阅有自己的进度。订阅方式决定同一订阅里的消费者怎样分担：

订阅类型怎样分担适用取舍
Exclusive一个订阅只允许一个消费者模型简单，消费者并发受限
Failover存在活动消费者与备用者；分区 Topic 可分担分区希望故障接手，同时保留相应分区顺序
Shared多个消费者分担消息，逐条确认吞吐优先，不承诺消息处理顺序
Key_Shared按 key / ordering key 分配要单键有序又需要多个消费者；有组批和重投边界

Key_Shared 并不是给 Shared 换个名字。生产者要设置合适的 key，并禁用跨键混合的默认组批或采用 key-based batching；否则一个批次包含不同键会破坏期望路由。负确认与主动重投也可能改变原顺序。Shared / Key_Shared 不支持用累计确认替代所有逐条确认。[14]

还要区分 durable subscription 与 non-durable 读取。消息是否继续保留，受 backlog、retention、TTL、订阅存在与否等策略影响；有 Message ID 不等于那条记录会永久存在。多租户和 namespace 让隔离策略有明确位置，但租户数量大时，权限、配额和资源治理仍然是需要运维的实际工作。

如果需求包含多租户、不同订阅语义、较大持久积压，以及团队能够维护存储层，Pulsar 值得进入候选。若只是单个应用的几个后台任务，则需要先证明这些能力带来的收益足以抵消组件数量和故障排查成本。

Redis Streams：不要把 PEL 当成自动重试引擎 ​

Redis Stream 是 Redis 里的追加数据结构。它不同于 Redis Pub/Sub：条目保存在 Stream 中，有 ID，可供读取和回放；持久恢复能力仍取决于 Redis 的 AOF/RDB、复制与故障恢复配置。一个 Stream key 也不会因为用了 Redis Cluster 就自动拆成多个分区。[15][16][17]

最常用的操作顺序如下：

命令做什么容易漏掉的边界
XADD追加字段和值，得到 Stream IDID 是存储位置，业务去重仍需事件 ID
XGROUP CREATE建立消费组，选择起始位置从头消费和只读新消息是不同业务决定
XREADGROUP ... >为该组读取尚未投递的新消息默认将消息记入 PEL，不会自动代表业务完成
XPENDING查看尚未确认的条目队列没有新消息，不代表 PEL 已处理完
XAUTOCLAIM按空闲时间接管 pending 条目必须处理返回游标，旧 worker 的业务请求未必已停止
XACK从该消费组的 PEL 移除确认记录不删除 Stream 本体中的条目
XTRIM / XDEL裁剪或删除日志内容要和消费滞后、PEL、保留目标一起设计

这些命令各有版本演进；XAUTOCLAIM 自 Redis 6.2 提供，本文没有使用较新版本才出现的组合命令或选项。NOACK 会跳过 pending 跟踪，不适合直接代替可靠工作流的确认。[18][19][20][21]

假设 worker A 取到一条消息后停住。仅仅不断用 > 读新消息，不能当作已经完成历史 pending 的恢复流程。需要扫描/接管待确认记录，设置合理空闲阈值，并保证重复执行安全。下面用模型演示“未到阈值不能接管，到阈值后更换 owner，ack 只清 pending”：

go
s := Stream{entries: []string{"event-1"}}
s.Deliver(0, "worker-a", 0)
fmt.Println("claim too soon:", s.Claim(0, "worker-b", 3, 5))
fmt.Println("claim after idle:", s.Claim(0, "worker-b", 8, 5))
fmt.Println("new owner:", s.pending[0].Owner)
s.Ack(0)
fmt.Println("pending, stored:", len(s.pending), len(s.entries))
text
claim too soon: false
claim after idle: true
new owner: worker-b
pending, stored: 0 1

这段代码没有 Redis 连接，省略了 Stream ID、PEL 游标、删除条目和网络并发等机制。它尤其没有终止 worker A，真实系统也不能用“消息已被 claim”推断旧 worker 的外部副作用已取消。数据库幂等或所有权版本检查依然需要。

Redis Streams 的优势是可以复用现有 Redis 运维与客户端，并在数据结构层组合流程。缺点是任务重试、退避、最大次数、隔离和可视化管理常需自己补；与缓存共用实例时，内存压力、驱逐和流量峰值会互相影响。若这些需求越来越多，成熟任务队列库可能比继续手写命令循环更合适。

怎样把候选缩到能做验证的两种 ​

不要先做所有产品的吞吐排行榜。先写一段能失败的业务场景，例如“订单提交，生产者在确认前断线，消费者在数据库提交后退出，十分钟后恢复”。要求候选方案逐项说明：事件在哪里、怎样恢复、会不会重复、如何定位坏消息、回放会不会再次产生外部效果。

然后再测工作负载：消息大小与峰值、可接受延迟、保留时长、消费组数量、热点键、故障域和运维能力。路由规则是主角时看 RabbitMQ；生产交接和定时消息是主角时评估 RocketMQ；已有 NATS 且要补持久化时看 JetStream；多租户与存储拆分是关键时看 Pulsar；需求有限且 Redis 边界可接受时看 Streams。它们都是带条件的候选理由，不是无需测量的最终排名。

依据与验证边界 ​

教学程序在 Go 1.27.2、Linux/amd64 执行，go run、测试、race、vet 均通过。测试包括 RabbitMQ # 位于中间、* 不能匹配零段，以及 NATS > 仅位于末尾且至少匹配一段的模型行为。没有服务器集成、网络故障、持久化或性能实验，不能据此声称六种产品已完成选型验证。

完整可复现代码 ​

将以下文件放在同一目录，执行：

sh
go run .
go test -v ./...
go test -race ./...
go vet ./...

go.mod ​

go.mod
module example.com/mq/choices

go 1.27.2

main.go ​

代码块 · 75 行 · 点击收起 / 展开
go
package main

import (
	"fmt"
	"strings"
)

// These matchers model nonempty dot-separated tokens, not full broker validation.
func rabbitMatch(pattern, key string) bool {
	p, k := strings.Split(pattern, "."), strings.Split(key, ".")
	var match func(int, int) bool
	match = func(i, j int) bool {
		if i == len(p) {
			return j == len(k)
		}
		if p[i] == "#" {
			return match(i+1, j) || (j < len(k) && match(i, j+1))
		}
		return j < len(k) && (p[i] == "*" || p[i] == k[j]) && match(i+1, j+1)
	}
	return match(0, 0)
}

func natsMatch(pattern, subject string) bool {
	p, s := strings.Split(pattern, "."), strings.Split(subject, ".")
	for i, token := range p {
		if token == ">" {
			return i == len(p)-1 && i < len(s)
		}
		if i >= len(s) || (token != "*" && token != s[i]) {
			return false
		}
	}
	return len(p) == len(s)
}

type Pending struct {
	Owner        string
	LastDelivery int
}
type Stream struct {
	entries []string
	pending map[int]Pending
}

func (s *Stream) Deliver(index int, owner string, now int) {
	if s.pending == nil {
		s.pending = make(map[int]Pending)
	}
	s.pending[index] = Pending{owner, now}
}
func (s *Stream) Claim(index int, owner string, now, minIdle int) bool {
	p, ok := s.pending[index]
	if !ok || now-p.LastDelivery < minIdle {
		return false
	}
	s.pending[index] = Pending{owner, now}
	return true
}
func (s *Stream) Ack(index int) { delete(s.pending, index) }

func demo() {
	fmt.Println("rabbit order.# / order:", rabbitMatch("order.#", "order"))
	fmt.Println("nats order.> / order:", natsMatch("order.>", "order"))
	fmt.Println("rabbit order.* / order.paid:", rabbitMatch("order.*", "order.paid"))
	fmt.Println("nats order.* / order.paid.eu:", natsMatch("order.*", "order.paid.eu"))
	s := Stream{entries: []string{"event-1"}}
	s.Deliver(0, "worker-a", 0)
	fmt.Println("claim too soon:", s.Claim(0, "worker-b", 3, 5))
	fmt.Println("claim after idle:", s.Claim(0, "worker-b", 8, 5))
	fmt.Println("new owner:", s.pending[0].Owner)
	s.Ack(0)
	fmt.Println("pending, stored:", len(s.pending), len(s.entries))
}
func main() { demo() }

main_test.go ​

go
package main

import "testing"

func Example() {
	demo()
	// Output:
	// rabbit order.# / order: true
	// nats order.> / order: false
	// rabbit order.* / order.paid: true
	// nats order.* / order.paid.eu: false
	// claim too soon: false
	// claim after idle: true
	// new owner: worker-b
	// pending, stored: 0 1
}
func TestRabbitHashInMiddle(t *testing.T) {
	for _, key := range []string{"a.b", "a.x.b", "a.x.y.b"} {
		if !rabbitMatch("a.#.b", key) {
			t.Fatal(key)
		}
	}
	if rabbitMatch("a.*.b", "a.b") {
		t.Fatal("star matched zero tokens")
	}
}
func TestNatsTailRequiresToken(t *testing.T) {
	if natsMatch("a.>", "a") || !natsMatch("a.>", "a.b.c") || natsMatch("a.>.b", "a.x.b") {
		t.Fatal("bad tail wildcard")
	}
}

  1. RabbitMQ 4.2 Exchanges：交换机类型、binding、默认交换机。 ↩︎ ↩︎

  2. RabbitMQ 4.2 Confirms：confirm、ack、return 与 prefetch。 ↩︎

  3. RabbitMQ 4.2 Quorum Queues 与 Streams：复制队列、已移除的 classic mirroring 和日志模型。 ↩︎

  4. RocketMQ 5.0 Transaction Message：半消息、二阶段结果、回查、消费侧边界与超时。 ↩︎ ↩︎ ↩︎

  5. RocketMQ 5.0 Delay Message：时间戳与使用限制。 ↩︎

  6. NATS JetStream 概念页对 Core NATS 与 JetStream 的区分。 ↩︎

  7. NATS Subjects：分段与通配符。 ↩︎

  8. JetStream Concepts：stream、consumer、持久化与重投。 ↩︎

  9. NATS Pull consumers in depth：Fetch 与 Consume 的后台拉取行为。 ↩︎

  10. NATS Retention policies。 ↩︎

  11. NATS 官方文档源码，JetStream Model Deep Dive：去重窗口与双确认；滚动资料，以查阅日期为界。 ↩︎

  12. NATS 官方文档源码，Consumers：ack、MaxDeliver 与 advisory。 ↩︎

  13. Pulsar 4.0.x Architecture Overview：Broker、BookKeeper 与元数据。 ↩︎

  14. Pulsar 4.0.x Messaging：四种订阅、Key_Shared 组批与确认限制、保留和游标。 ↩︎

  15. Redis XADD 与 XGROUP CREATE：追加和消费组起点。 ↩︎

  16. Redis Cluster specification：key 与 hash slot。 ↩︎

  17. Redis persistence：AOF/RDB 与各自恢复边界。 ↩︎

  18. Redis XREADGROUP 与 XPENDING：PEL、> 和 NOACK。 ↩︎

  19. Redis XACK：移除 PEL 记录而非 Stream 内容。 ↩︎

  20. Redis XAUTOCLAIM：版本、空闲阈值、游标及接管。 ↩︎

  21. Redis XTRIM：裁剪及版本相关选项。 ↩︎

保留各书来源与许可 · 示例运行条件以正文为准