03 · Pub/Sub 发布订阅
NATS 系列指南第 3 篇 — 深入理解最基础的消息传递模式
目录
- Pub/Sub 基本概念
- 消息结构详解
- 订阅类型
- 扇出（Fan-out）机制
- 消息传递语义：At Most Once
- 内部实现原理
- Slow Consumer 问题
- Headers 使用（NATS 2.2+）
- 实际代码示例（Go）
- Core Pub/Sub 局限性
- 小结
1. Pub/Sub 基本概念
Pub/Sub（发布/订阅） 是 NATS 最基础的通信模式，也是整个消息系统的核心。它将消息的发送方（Publisher） 和接收方（Subscriber） 完全解耦，双方不需要知道对方的存在，只通过 Subject 作为媒介进行通信。
1.1 三个核心特性
解耦（Decoupling）
传统直连模式（紧耦合）:
Service A ──直接调用──▶ Service B
──直接调用──▶ Service C
──直接调用──▶ Service D
Service A 必须知道 B、C、D 的地址，任何一个宕机都影响 A
Pub/Sub 模式（松耦合）:
Service A
│ PUB "order.created"
▼
[NATS Server]
│
├──▶ Service B（SUB "order.created"）
├──▶ Service C（SUB "order.created"）
└──▶ Service D（SUB "order.*"）
Service A 只需发布到 Subject，不关心谁在监听异步（Asynchronous）
发布者发布消息后立即返回，不等待订阅者处理完成。订阅者在各自的时间节奏中处理消息，两者运行节奏完全独立。
一对多（One-to-Many / Fan-out）
单次发布的一条消息可以同时被任意数量的订阅者接收，这是 Pub/Sub 区别于点对点（Queue）模式的根本特征。
1.2 与 Request/Reply 和 Queue Group 的对比
| 模式 | 发布者 | 订阅者 | 典型用途 |
|---|---|---|---|
| Pub/Sub | 1 | N（所有匹配者） | 事件广播、通知 |
| Request/Reply | 1 | 1（第一个响应者） | RPC 调用 |
| Queue Group | 1 | 1（组内随机一个） | 负载均衡 |
2. 消息结构详解
NATS 消息（*nats.Msg）包含以下字段：
┌────────────────────────────────────────────────┐
│ NATS Message │
│ │
│ Subject: "orders.created" │
│ Reply-To: "_INBOX.4k9fVa3Km1G..." (可选) │
│ │
│ Headers: (NATS 2.2+, 可选) │
│ Nats-Msg-Id: "uuid-1234" │
│ Content-Type: "application/json" │
│ X-Custom-Header: "value" │
│ │
│ Payload: (字节数组，应用自定义格式) │
│ {"order_id":"ORD-001","amount":99.00} │
│ │
└────────────────────────────────────────────────┘2.1 Subject
消息的路由地址，决定哪些订阅者能收到此消息。详见 02 · Subject 寻址系统。
2.2 Reply-To
可选字段，由发布者填写，用于 Request/Reply 模式中指定回复地址（通常是 _INBOX.<token>）。普通 Pub/Sub 场景中此字段为空。
2.3 Headers（NATS 2.2+）
类似 HTTP Headers，是键值对形式的元数据。Headers 不影响路由，但可以携带结构化的上下文信息：
Nats-Msg-Id: <unique-id> # 消息去重 ID（JetStream 使用）
Nats-Expected-Stream: ORDERS # 期望写入的 Stream 名（校验用）
Nats-Expected-Last-Sequence: 42 # 期望的上一条序号（乐观并发控制）
Content-Type: application/json # 消息体格式标注
X-Trace-Id: <span-id> # 链路追踪 ID2.4 Payload
消息正文，是纯字节数组（[]byte）。NATS 对格式不做任何假设，可以是：
- JSON：
{"event":"order_created","id":"001"} - Protobuf 序列化字节
- MessagePack 字节
- 纯文本字符串
- 空（zero-byte payload，仅用 Subject 传递信号）
消息大小限制：默认最大 1 MB，可通过服务器配置 max_payload 调整（最大 64 MB）。
2.5 NATS 协议层面的消息格式
NATS 使用文本协议，发布消息的原始格式为：
PUB <subject> [reply-to] <#bytes>\r\n
[payload]\r\n
# 示例（无 reply-to，payload 为 JSON）
PUB orders.created 37\r\n
{"order_id":"001","amount":99.00}\r\n
# 带 Headers 的格式（NATS 2.2+）
HPUB <subject> [reply-to] <#header bytes> <#total bytes>\r\n
NATS/1.0\r\n
Nats-Msg-Id: abc123\r\n
\r\n
[payload]\r\n3. 订阅类型
3.1 同步订阅 vs 异步订阅
异步订阅（推荐）
使用回调函数，消息到达时由 NATS 客户端内部的 goroutine 自动调用：
go
// 异步订阅：注册回调，非阻塞
sub, err := nc.Subscribe("orders.created", func(msg *nats.Msg) {
// 此回调在 NATS 客户端的分发 goroutine 中执行
var order Order
json.Unmarshal(msg.Data, &order)
processOrder(order)
})
// Subscribe 立即返回，程序继续执行其他逻辑同步订阅
使用 SubscribeSync，需要主动调用 NextMsg 来阻塞等待消息：
go
// 同步订阅：需要主动拉取
sub, err := nc.SubscribeSync("orders.created")
// 阻塞等待，最多等待 5 秒
msg, err := sub.NextMsg(5 * time.Second)
if err == nats.ErrTimeout {
fmt.Println("没有消息到来")
return
}
processMessage(msg)对比：
| 特性 | 异步订阅 | 同步订阅 |
|---|---|---|
| 使用场景 | 持续监听，生产环境主流 | 测试、脚本、精确控制处理时机 |
| 阻塞行为 | 不阻塞，回调由内部 goroutine 触发 | 阻塞直到消息到来或超时 |
| 并发处理 | 默认串行（单个 subscription 的回调不并发） | 完全由调用方控制 |
| 错误处理 | 在回调内处理，panic 会被捕获并断开连接 | 正常的 Go 错误返回 |
3.2 普通订阅 vs 通配符订阅
go
// 精确匹配订阅
nc.Subscribe("orders.created", handler)
// 单层通配符订阅
nc.Subscribe("orders.*", handler) // 匹配 orders.created, orders.cancelled 等
// 多层通配符订阅
nc.Subscribe("orders.>", handler) // 匹配所有 orders. 开头的 Subject
// 多个通配符组合
nc.Subscribe("iot.*.sensor.*", handler) // 匹配 iot.factoryA.sensor.temp 等通配符订阅在回调中可以通过 msg.Subject 获取实际收到的精确 Subject：
go
nc.Subscribe("orders.*", func(msg *nats.Msg) {
fmt.Printf("收到消息，实际 Subject: %s\n", msg.Subject)
// 输出示例: "收到消息，实际 Subject: orders.created"
// 解析具体事件类型
parts := strings.Split(msg.Subject, ".")
eventType := parts[len(parts)-1] // "created", "cancelled", etc.
switch eventType {
case "created":
handleOrderCreated(msg.Data)
case "cancelled":
handleOrderCancelled(msg.Data)
}
})3.3 一次性订阅（AutoUnsubscribe）
使用 AutoUnsubscribe 可以让订阅在收到指定数量的消息后自动取消：
go
// 只接收 1 条消息后自动取消订阅
sub, _ := nc.Subscribe("orders.created", handler)
sub.AutoUnsubscribe(1)
// 链式写法：只接收 3 条
sub, _ := nc.Subscribe("alerts.>", func(msg *nats.Msg) {
fmt.Printf("收到告警: %s\n", string(msg.Data))
})
sub.AutoUnsubscribe(3)AutoUnsubscribe 常用于：
- Request/Reply 的请求方：发出请求后只期望一个回复
- 启动时的一次性初始化数据获取
- 测试场景中精确控制消息消费数量
3.4 取消订阅
go
// 立即取消
sub.Unsubscribe()
// 再收 N 条后取消（等同于 AutoUnsubscribe）
sub.AutoUnsubscribe(N)
// 带超时的订阅（需要结合 context 或 timer 手动控制）
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
sub, _ := nc.Subscribe("events.>", handler)
<-ctx.Done()
sub.Unsubscribe()4. 扇出（Fan-out）机制
4.1 什么是扇出
扇出是指一条发布的消息被同时投递给所有匹配的订阅者。NATS 服务器在收到发布消息后，会立即查找所有匹配的订阅，并并行地将消息副本推送给每一个订阅者。
发布: PUB "orders.created" msg
│
▼
┌─────────────────┐
│ NATS Server │
│ │
│ 查找匹配订阅: │
│ ✓ orders.created (Client B)
│ ✓ orders.* (Client C)
│ ✓ orders.> (Client D)
│ ✗ payments.* (不匹配) │
└────────┬────────┘
│
┌──────────────┼──────────────┐
│ │ │
▼ ▼ ▼
Client B Client C Client D
(orders.created) (orders.*) (orders.>)4.2 扇出的实际应用
事件驱动架构中的典型扇出场景：
PUB "orders.created" {order_id: "001", user_id: "U123", amount: 299.00}
│
├──▶ 库存服务 → 锁定库存
├──▶ 通知服务 → 发送确认邮件
├──▶ 风控服务 → 记录交易行为
├──▶ 数据分析服务 → 更新实时报表
└──▶ 积分服务 → 计算奖励积分所有这些服务同时收到消息，独立处理，互不阻塞，发布方无感知。
4.3 扇出与 Queue Group 的区别
Pub/Sub 扇出（Fan-out）:
PUB "orders.created" msg
│
├──▶ 订阅者 A （收到）
├──▶ 订阅者 B （收到）
└──▶ 订阅者 C （收到）
每个订阅者都收到一份消息副本
Queue Group（竞争消费）:
PUB "orders.created" msg
│
▼
Queue Group "processors"
│
├── 实例 1 ← 随机选一个（收到）
├── 实例 2 （不收）
└── 实例 3 （不收）
只有一个实例收到消息（负载均衡）实际使用建议：
- 如果每个服务类型都需要收到消息 → Pub/Sub 扇出
- 如果同一服务类型有多个实例，希望负载均衡 → Queue Group
5. 消息传递语义：At Most Once
5.1 Core NATS 的语义保证
Core NATS（不使用 JetStream）采用 At Most Once（最多一次） 语义：
- 消息最多被投递一次
- 不保证消息一定被送达
- 不重传失败的消息
- 不持久化消息（内存缓冲溢出则丢弃）
At Most Once 的含义:
场景 1: 正常投递
Publisher ──msg──▶ Server ──msg──▶ Subscriber ✓ 投递一次
场景 2: 订阅者不存在
Publisher ──msg──▶ Server 订阅者不在线，消息丢弃 ✗
场景 3: 订阅者处理太慢（Slow Consumer）
Publisher ──大量msg──▶ Server
Server buffer 溢出
丢弃部分消息 ✗
向订阅者发送 Slow Consumer 警告
场景 4: 网络抖动
Publisher ──msg──▶ [网络断开] 消息丢失 ✗5.2 为什么不是 At Least Once
NATS 有意选择 At Most Once 作为 Core 层的语义，原因是：
- 极低延迟：不需要 ACK 机制，发完即走
- 高吞吐：服务器无需维护消息状态
- 简单可靠：适合实时性要求高但允许偶发丢失的场景（如监控指标、日志流）
对于需要At Least Once 或 Exactly Once 的场景，应使用 JetStream。
5.3 适合 Core Pub/Sub 的场景
| 场景 | 原因 |
|---|---|
| 实时监控指标上报 | 偶发丢失可接受，最新值比历史更重要 |
| 日志流传输 | 少量丢失可接受，吞吐量优先 |
| 在线用户状态广播 | 状态会持续更新，丢一次下次会补上 |
| 游戏位置同步 | 实时性优先，旧位置数据价值低 |
| 心跳/存活检测 | 单次丢失不影响逻辑 |
6. 内部实现原理
6.1 服务器端订阅表（Interest Map）
NATS 服务器内部维护一个兴趣映射表（interest map），记录每个 Subject 对应哪些客户端连接正在订阅。
Interest Map 示意（简化）:
┌─────────────────────────────────────────────────────┐
│ Interest Map │
│ │
│ "orders.created" → [conn-1, conn-3, conn-7] │
│ "orders.*" → [conn-2] │
│ "orders.>" → [conn-4, conn-5] │
│ "payments.>" → [conn-6] │
│ "iot.*.sensor.*" → [conn-8] │
│ │
└─────────────────────────────────────────────────────┘
实际实现使用前缀树（Subject Trie）而非哈希表，
以支持高效的通配符匹配。6.2 消息分发流程
完整的消息分发流程:
Step 1: 发布者发送消息
┌──────────┐
│ Publisher│ PUB "orders.created" 37\r\n{...payload...}
└────┬─────┘
│ TCP/TLS
▼
┌────────────────────────────────────────────────────┐
│ NATS Server │
│ │
│ Step 2: Parser 解析协议帧 │
│ ┌─────────────┐ │
│ │ Protocol │ 解析 Subject、Reply-To、Payload │
│ │ Parser │ │
│ └──────┬──────┘ │
│ │ │
│ Step 3: Router 查找匹配订阅 │
│ ┌──────▼──────┐ │
│ │ Router/ │ 遍历 Subject Trie，找到所有匹配 │
│ │ Interest │ 的订阅者连接 │
│ │ Map │ │
│ └──────┬──────┘ │
│ │ 匹配到: conn-1, conn-2, conn-4 │
│ │ │
│ Step 4: 并发写入各连接的发送缓冲区 │
│ ┌──────▼──────────────────────────────────────┐ │
│ │ Write to outbound buffers │ │
│ │ │ │
│ │ conn-1 outbuf: [msg copy] │ │
│ │ conn-2 outbuf: [msg copy] │ │
│ │ conn-4 outbuf: [msg copy] │ │
│ └──────────────────────────────────────────┬──┘ │
│ │ │
└─────────────────────────────────────────────┼──────┘
Step 5: 异步 flush │
┌───────────────┬───────────────┘
▼ ▼
┌──────────┐ ┌──────────┐
│Subscriber│ │Subscriber│ 消息到达
│ conn-1 │ │ conn-2 │
└──────────┘ └──────────┘6.3 背压和流控：Slow Consumer 机制
每个订阅者连接在服务器端都有一个发送缓冲区（outbound buffer）。当订阅者处理速度慢于发布速度时：
时间轴:
t=0 Publisher 发布 msg-1, msg-2, msg-3, ...
Server outbuf for Subscriber:
t=1 [msg-1][msg-2][msg-3] 缓冲区开始积压
t=2 [msg-1][msg-2]...[msg-50] 缓冲区继续增长
t=3 [msg-1][msg-2]...[msg-100] 接近上限（默认 65536 消息 / 64MB）
t=4 缓冲区溢出！
→ 服务器记录 "Slow Consumer" 错误
→ 服务器开始丢弃该连接的新消息
→ 服务器可能断开该连接（配置相关）默认的 pending 上限（服务器端）：
- 消息数：65536 条
- 字节数：64 MB
超过任意一个上限，服务器将向客户端发送 -ERR 'Slow Consumer' 错误，并开始丢弃消息。
7. Slow Consumer 问题
7.1 什么是 Slow Consumer
Slow Consumer 是指消息消费速度持续低于消息产生速度的订阅者。这会导致：
- 服务器端缓冲区积压
- 消息开始丢失
- 严重时服务器主动断开连接
正常状态:
Publisher ─[100msg/s]─▶ NATS Server ─[100msg/s]─▶ Subscriber(处理100msg/s)
buffer: ~0
Slow Consumer 状态:
Publisher ─[1000msg/s]─▶ NATS Server ─[slow]─▶ Subscriber(处理10msg/s)
buffer: 持续增长 ↑↑↑ → 溢出 → 丢消息7.2 服务器端缓冲机制
服务器为每个客户端连接维护的发送缓冲区配置（nats-server.conf）：
# 服务器配置
max_pending = 67108864 # 单个连接最大 pending 字节数（64MB）
write_deadline = "2s" # 写超时时间，超过则断开客户端库层面的缓冲（Go 客户端）：
go
nc, err := nats.Connect("nats://localhost:4222",
// 设置客户端接收缓冲区大小（默认 32768 消息）
nats.SubPendingMsgsLimit(50000),
// 设置客户端接收缓冲字节上限（默认 65536000 字节 ~64MB）
nats.SubPendingBytesLimit(100*1024*1024), // 100MB
)7.3 如何避免和处理 Slow Consumer
方案 1：提高消费者处理速度
go
// 错误：回调中做耗时操作（阻塞分发 goroutine）
nc.Subscribe("orders.>", func(msg *nats.Msg) {
result := callDatabaseWithTimeout(msg.Data) // 可能阻塞 100ms+
sendHTTPRequest(result) // 可能阻塞 200ms+
})
// 正确：将耗时操作推入工作 goroutine
jobs := make(chan *nats.Msg, 10000)
// 启动工作池
for i := 0; i < 10; i++ {
go func() {
for msg := range jobs {
result := callDatabase(msg.Data)
sendHTTPRequest(result)
}
}()
}
nc.Subscribe("orders.>", func(msg *nats.Msg) {
// 回调只做非阻塞的入队操作
select {
case jobs <- msg:
default:
log.Warn("job queue full, dropping message")
}
})方案 2：使用 Queue Group 分散压力
go
// 同一服务启动多个实例，共享消费压力
nc.QueueSubscribe("orders.created", "order-processor-group", func(msg *nats.Msg) {
processOrder(msg.Data)
})
// 启动 10 个实例 = 消费能力 10x方案 3：使用 JetStream 实现背压
JetStream 的 Push Consumer 提供流控（Flow Control）机制，Consumer 可以通过发送 ACK 来控制服务端的推送速率，从根本上解决 Slow Consumer 问题。
方案 4：监控和告警
go
// 订阅 Slow Consumer 错误事件
nc.SetErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) {
if err == nats.ErrSlowConsumer {
pending, limit, _ := sub.Pending()
dropped, _ := sub.Dropped()
log.Errorf("Slow consumer on %s: pending=%d/%d, dropped=%d",
sub.Subject, pending, limit, dropped)
// 触发告警
alerting.Send("slow_consumer", map[string]interface{}{
"subject": sub.Subject,
"pending": pending,
"dropped": dropped,
})
}
})查看订阅的积压状态：
go
// 获取当前积压的消息数和字节数
msgs, bytes, err := sub.Pending()
fmt.Printf("积压消息: %d 条 / %d 字节\n", msgs, bytes)
// 获取因 Slow Consumer 丢弃的消息数
dropped, err := sub.Dropped()
fmt.Printf("已丢弃消息: %d 条\n", dropped)8. Headers 使用（NATS 2.2+）
8.1 Headers 基础
NATS 2.2 引入了消息 Headers，语法与 HTTP Headers 类似：
go
// 发布带 Headers 的消息
msg := &nats.Msg{
Subject: "orders.created",
Header: nats.Header{},
Data: []byte(`{"order_id":"001"}`),
}
msg.Header.Set("Content-Type", "application/json")
msg.Header.Set("X-Trace-Id", "span-abc-123")
msg.Header.Set("X-Source-Service", "order-service")
nc.PublishMsg(msg)go
// 订阅者读取 Headers
nc.Subscribe("orders.created", func(msg *nats.Msg) {
traceId := msg.Header.Get("X-Trace-Id")
contentType := msg.Header.Get("Content-Type")
fmt.Printf("TraceID: %s, ContentType: %s\n", traceId, contentType)
// 处理消息体
processPayload(msg.Data)
})8.2 JetStream 特殊 Headers
| Header 名 | 用途 | 示例值 |
|---|---|---|
Nats-Msg-Id | 消息幂等性去重 ID | uuid-1234-5678 |
Nats-Expected-Stream | 校验期望写入的 Stream | ORDERS |
Nats-Expected-Last-Sequence | 乐观并发控制：期望上一条 Seq | 42 |
Nats-Expected-Last-Msg-Id | 乐观并发控制：期望上一条 Msg-Id | uuid-prev |
Nats-Expected-Last-Subject-Sequence | 同一 Subject 的上一条 Seq | 10 |
消息去重示例（配合 JetStream）：
go
msg := &nats.Msg{
Subject: "orders.created",
Header: nats.Header{},
Data: payload,
}
// 设置幂等 ID，JetStream 会在去重窗口内忽略重复 ID
msg.Header.Set("Nats-Msg-Id", order.ID)
js.PublishMsg(msg)8.3 链路追踪集成
go
// 发布时注入 OpenTelemetry trace context
func publishWithTrace(ctx context.Context, nc *nats.Conn, subject string, data []byte) error {
msg := &nats.Msg{
Subject: subject,
Header: nats.Header{},
Data: data,
}
// 将 trace context 注入 headers
otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(msg.Header))
return nc.PublishMsg(msg)
}
// 订阅时提取 trace context
nc.Subscribe("orders.>", func(msg *nats.Msg) {
// 从 headers 提取 trace context
ctx := otel.GetTextMapPropagator().Extract(
context.Background(),
propagation.HeaderCarrier(msg.Header),
)
span, ctx := tracer.Start(ctx, "process-order")
defer span.End()
processOrderWithCtx(ctx, msg.Data)
})9. 实际代码示例（Go）
9.1 基础 Pub/Sub
go
package main
import (
"fmt"
"log"
"time"
"github.com/nats-io/nats.go"
)
func main() {
// 连接 NATS 服务器
nc, err := nats.Connect(nats.DefaultURL,
nats.Name("order-service"),
nats.ReconnectWait(2*time.Second),
nats.MaxReconnects(10),
nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
log.Printf("断开连接: %v", err)
}),
nats.ReconnectHandler(func(nc *nats.Conn) {
log.Printf("重新连接到: %s", nc.ConnectedUrl())
}),
)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
// ── 订阅方 ──
sub, err := nc.Subscribe("orders.created", func(msg *nats.Msg) {
fmt.Printf("[订阅者] 收到消息:\n")
fmt.Printf(" Subject: %s\n", msg.Subject)
fmt.Printf(" Payload: %s\n", string(msg.Data))
})
if err != nil {
log.Fatal(err)
}
defer sub.Unsubscribe()
// 确保订阅已经到达服务器
nc.Flush()
// ── 发布方 ──
err = nc.Publish("orders.created", []byte(`{"order_id":"ORD-001","amount":99.00}`))
if err != nil {
log.Fatal(err)
}
fmt.Println("[发布者] 消息已发布")
// 等待消息处理
time.Sleep(100 * time.Millisecond)
}9.2 通配符订阅
go
package main
import (
"fmt"
"log"
"strings"
"time"
"github.com/nats-io/nats.go"
)
func main() {
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
// 通配符订阅：匹配所有 orders 下的事件
nc.Subscribe("orders.>", func(msg *nats.Msg) {
// 从 Subject 中提取事件类型
parts := strings.Split(msg.Subject, ".")
domain := parts[0]
event := strings.Join(parts[1:], ".")
fmt.Printf("[orders.>] domain=%s event=%s payload=%s\n",
domain, event, string(msg.Data))
})
// 另一个通配符：只匹配第二层
nc.Subscribe("orders.*", func(msg *nats.Msg) {
fmt.Printf("[orders.*] subject=%s\n", msg.Subject)
})
// 精确订阅（优先级最高，但 NATS 不去重，会同时触发所有匹配订阅）
nc.Subscribe("orders.created", func(msg *nats.Msg) {
fmt.Printf("[orders.created] 精确匹配: %s\n", string(msg.Data))
})
nc.Flush()
// 发布测试消息
testMessages := []struct {
subject string
payload string
}{
{"orders.created", `{"id":"001"}`}, // 触发全部 3 个订阅
{"orders.cancelled", `{"id":"002"}`}, // 触发 orders.> 和 orders.*
{"orders.payment.failed", `{"id":"003"}`}, // 只触发 orders.>
}
for _, tm := range testMessages {
fmt.Printf("\n── 发布 %s ──\n", tm.subject)
nc.Publish(tm.subject, []byte(tm.payload))
time.Sleep(50 * time.Millisecond)
}
time.Sleep(200 * time.Millisecond)
}9.3 扇出示例
go
package main
import (
"fmt"
"log"
"sync"
"time"
"github.com/nats-io/nats.go"
)
// 模拟不同的下游服务
func inventoryService(nc *nats.Conn, wg *sync.WaitGroup) {
nc.Subscribe("orders.created", func(msg *nats.Msg) {
fmt.Printf("[库存服务] 收到订单，开始锁定库存: %s\n", string(msg.Data))
time.Sleep(10 * time.Millisecond) // 模拟处理
fmt.Println("[库存服务] 库存已锁定")
wg.Done()
})
}
func notificationService(nc *nats.Conn, wg *sync.WaitGroup) {
nc.Subscribe("orders.created", func(msg *nats.Msg) {
fmt.Printf("[通知服务] 收到订单，发送确认邮件: %s\n", string(msg.Data))
time.Sleep(20 * time.Millisecond) // 模拟发邮件
fmt.Println("[通知服务] 确认邮件已发送")
wg.Done()
})
}
func analyticsService(nc *nats.Conn, wg *sync.WaitGroup) {
nc.Subscribe("orders.>", func(msg *nats.Msg) {
fmt.Printf("[分析服务] 记录事件: subject=%s\n", msg.Subject)
wg.Done()
})
}
func main() {
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
var wg sync.WaitGroup
// 启动三个"服务"（模拟独立服务订阅）
inventoryService(nc, &wg)
notificationService(nc, &wg)
analyticsService(nc, &wg)
nc.Flush()
// 每发布一条 orders.created，3 个服务都会收到
wg.Add(3)
payload := `{"order_id":"ORD-100","user_id":"U-456","amount":299.00}`
fmt.Printf("\n── 发布 orders.created ──\n发布方: %s\n\n", payload)
nc.Publish("orders.created", []byte(payload))
// 等待所有服务处理完成
wg.Wait()
fmt.Println("\n── 所有服务处理完成（扇出完成）──")
}9.4 带工作池的高吞吐消费者
go
package main
import (
"fmt"
"log"
"runtime"
"time"
"github.com/nats-io/nats.go"
)
type Worker struct {
id int
jobs <-chan *nats.Msg
}
func (w *Worker) start() {
go func() {
for msg := range w.jobs {
// 模拟处理
time.Sleep(5 * time.Millisecond)
fmt.Printf("[Worker %d] 处理消息: subject=%s\n", w.id, msg.Subject)
}
}()
}
func main() {
nc, err := nats.Connect(nats.DefaultURL,
// 设置更大的客户端缓冲区
nats.SubPendingMsgsLimit(100000),
nats.SubPendingBytesLimit(256*1024*1024),
)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
// 工作队列
numWorkers := runtime.NumCPU()
jobs := make(chan *nats.Msg, numWorkers*100)
// 启动工作池
for i := 0; i < numWorkers; i++ {
w := &Worker{id: i, jobs: jobs}
w.start()
}
var dropped int64
// 订阅者：仅做非阻塞入队
sub, err := nc.Subscribe("orders.>", func(msg *nats.Msg) {
select {
case jobs <- msg:
// 成功入队
default:
// 队满，记录丢弃（此处可以发告警）
dropped++
if dropped%100 == 0 {
log.Printf("警告：已丢弃 %d 条消息（消费者压力过大）", dropped)
}
}
})
if err != nil {
log.Fatal(err)
}
defer sub.Unsubscribe()
nc.Flush()
fmt.Printf("启动 %d 个 Worker，监听 orders.>\n", numWorkers)
// 模拟高频发布
go func() {
for i := 0; i < 1000; i++ {
nc.Publish("orders.created", []byte(fmt.Sprintf(`{"id":%d}`, i)))
}
fmt.Println("发布完成（1000 条消息）")
}()
time.Sleep(3 * time.Second)
// 打印统计
pending, _, _ := sub.Pending()
dropped2, _ := sub.Dropped()
fmt.Printf("\n统计: pending=%d, dropped_by_server=%d, dropped_by_us=%d\n",
pending, dropped2, dropped)
}9.5 AutoUnsubscribe 一次性订阅
go
package main
import (
"fmt"
"log"
"time"
"github.com/nats-io/nats.go"
)
func main() {
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
received := make(chan string, 10)
// 只接收前 3 条消息，之后自动取消订阅
sub, err := nc.Subscribe("notifications.>", func(msg *nats.Msg) {
received <- fmt.Sprintf("收到: %s -> %s", msg.Subject, string(msg.Data))
})
if err != nil {
log.Fatal(err)
}
// 设置只接收 3 条
sub.AutoUnsubscribe(3)
nc.Flush()
// 发布 5 条消息
for i := 1; i <= 5; i++ {
nc.Publish(fmt.Sprintf("notifications.msg-%d", i),
[]byte(fmt.Sprintf("消息 #%d", i)))
time.Sleep(10 * time.Millisecond)
}
// 只会收到 3 条
time.Sleep(100 * time.Millisecond)
close(received)
fmt.Println("接收到的消息（应该只有 3 条）:")
for msg := range received {
fmt.Println(" ", msg)
}
fmt.Printf("订阅是否有效: %v\n", sub.IsValid())
}10. Core Pub/Sub 局限性
10.1 无持久化
订阅者不在线时发布的消息会永久丢失：
t=0 订阅者下线（崩溃/重启）
t=1 发布者发布 1000 条消息 → 全部丢失
t=2 订阅者重新上线
t=3 订阅者收不到 t=1 期间的任何消息10.2 无重传
消息投递失败（网络中断、Slow Consumer 丢弃）后不会自动重传：
网络抖动期间丢失的消息:
t=0 连接正常
t=1 [网络抖动 500ms]
发布的消息全部丢失，无重传
t=2 连接恢复
发布的新消息正常投递10.3 无消息顺序保证（多订阅者场景）
同一消息广播给多个订阅者时，各订阅者处理完成的顺序不确定（受网络、CPU 调度影响）。
10.4 无背压机制
发布者可以无限制地发布消息，服务器对慢速消费者的唯一处理手段是丢弃消息并断开连接，而不是"暂停"发布者。
10.5 何时应该使用 JetStream
| 需求 | Core Pub/Sub | JetStream |
|---|---|---|
| 订阅者离线时不丢消息 | ✗ | ✓ |
| 消息重传（至少一次） | ✗ | ✓ |
| 精确一次（Exactly Once） | ✗ | ✓ |
| 消息历史回放 | ✗ | ✓ |
| 消息确认（ACK） | ✗ | ✓ |
| 流控/背压 | ✗ | ✓ |
| 消息过期（TTL） | ✗ | ✓ |
| 实时广播（在线订阅者） | ✓ | ✓ |
| 超低延迟（< 1ms） | ✓ | ~1ms |
| 最简单的使用方式 | ✓ | 需要配置 Stream |
11. 小结
Core Pub/Sub 全景图
发布者 NATS Server 订阅者
│ │ ┌────────────────┐
│ PUB subject msg │ │订阅1: subject │
│─────────────────────▶│ │订阅2: subject.*│
│ │ 查兴趣表 │订阅3: subject.>│
│ │ 找到所有匹配者 └────────────────┘
│ │ │
│ │──────────────────────────▶ 所有匹配者同时收到
│ │ （At Most Once，内存缓冲）
│ │
│ │ 注意：
│ │ - 无持久化
│ │ - 无 ACK
│ │ - 无重传
│ │ - Slow Consumer → 丢消息| 知识点 | 核心要点 |
|---|---|
| 基本概念 | 解耦、异步、一对多（扇出） |
| 消息结构 | Subject + Reply-To + Headers + Payload |
| 订阅类型 | 异步（推荐）/ 同步 / 通配符 / AutoUnsubscribe |
| 扇出 | 一条消息同时被所有匹配订阅者收到，与 Queue Group 互补 |
| 传递语义 | At Most Once，适合实时场景，不适合关键业务 |
| Slow Consumer | 消费慢 → 服务端缓冲积压 → 丢消息，解决方案：工作池、Queue Group、JetStream |
| Headers | NATS 2.2+ 支持，用于元数据、链路追踪、幂等控制 |
| 局限性 | 无持久化、无重传、无背压，复杂场景用 JetStream |
Core Pub/Sub 是 NATS 最纯粹、最高效的通信模式，理解它的能力边界，才能在正确的场景选择正确的工具。下一篇将介绍 Request/Reply 模式，展示如何在 Pub/Sub 基础上构建同步 RPC 语义。