04 - Request/Reply 与 Queue Groups
本文是 NATS 指南系列第四篇，深入讲解 Request/Reply（请求/应答）模式和 Queue Groups（队列订阅组）。这两个特性组合在一起，构成了 NATS 实现 RPC 服务调用与水平扩展的核心机制。
目录
- Request/Reply 概念
- NATS 如何实现 Request/Reply
- 完整交互时序图
- 超时处理
- 多应答模式
- 服务端负载模式
- 错误处理
- Go 代码示例：完整 Request/Reply
- Queue Groups 概念
- Queue Groups vs 普通订阅
- 分发策略
- Queue Groups 实际应用场景
- Go 代码示例：Queue Groups Worker Pool
- Queue Subscribe vs JetStream Consumer
- 实际架构示例：API Gateway + Queue Group + Service
- 小结
1. Request/Reply 概念
1.1 同步 RPC over 异步消息
传统的 HTTP RPC 是一种强同步模型：调用方阻塞等待服务端响应，双方通过 TCP 连接直接耦合。这种方式简单直观，但在高并发、服务发现、多实例场景下会面临挑战。
NATS 的 Request/Reply 实现了一种异步消息传输层之上的逻辑同步调用模式：
传统 HTTP RPC（强耦合）:
Client ───── TCP 连接 ─────▶ Server
◀─── 响应 (同一连接) ───
NATS Request/Reply（松耦合）:
Client ──publish──▶ NATS Broker ──route──▶ Server
◀──publish── ◀──publish──两者在调用方看起来都是"发送请求，等待响应"，但底层实现截然不同：
| 特性 | HTTP RPC | NATS Request/Reply |
|---|---|---|
| 传输层 | TCP 直连（需要知道服务地址） | 通过 NATS Broker 路由（无需知道地址） |
| 服务发现 | 需要额外机制（DNS/注册中心） | 天然内置（订阅 subject 即注册） |
| 多实例负载均衡 | 需要 Load Balancer | Queue Groups 内置支持 |
| 超时语义 | 连接超时 / 读超时 | 消息级别超时 |
| 单向 vs 双向 | 请求-响应（单次） | 可扩展为流式（多个 chunk） |
| 调用者耦合度 | 需要知道服务的网络地址 | 只需要知道 subject 名称 |
1.2 核心思想
Request/Reply 的本质是：请求方在发布消息时附上一个"回信地址"（reply-to subject），服务方收到消息后，将应答发布到该地址。
这个"回信地址"就是 _INBOX，NATS 为每个请求自动生成一个唯一的 inbox subject，使得回复能精确路由回发起方。
2. NATS 如何实现 Request/Reply
2.1 _INBOX 机制详解
当你调用 nc.Request("service.greet", data, timeout) 时，NATS 客户端在内部执行以下步骤：
步骤 1：生成唯一 Inbox
_INBOX.<nuid>
例如：_INBOX.4k9VJyEYEP1k1BfO2jnuid 是 NATS 自研的高性能唯一 ID 生成器（比 UUID v4 快约 30 倍），基于随机 base62 字符串，在单进程内全局唯一。
步骤 2：订阅该 Inbox
客户端在发送请求之前，先在本地订阅 _INBOX.4k9VJyEYEP1k1BfO2j（或更精确地，订阅 _INBOX.<prefix>.* 以复用连接上的单个订阅，提升性能）。
步骤 3：发布带有 ReplyTo 的消息
NATS 协议帧（简化）：
PUB service.greet _INBOX.4k9VJyEYEP1k1BfO2j 13\r\n
{"name":"world"}\r\n
^ ^
目标 subject reply-to subject步骤 4：服务端处理并应答
服务端收到消息后，提取 reply-to（即 _INBOX.4k9VJyEYEP1k1BfO2j），然后：
PUB _INBOX.4k9VJyEYEP1k1BfO2j "" 20\r\n
{"greeting":"Hello!!"}\r\n步骤 5：应答路由回客户端
NATS Broker 将该 PUB 路由到订阅了 _INBOX.4k9VJyEYEP1k1BfO2j 的客户端（即发起请求的那个连接），完成闭环。
2.2 Inbox 前缀优化（v2.2+）
早期版本每次 Request 都创建一个新订阅，高并发下会产生大量的 SUB/UNSUB 协议帧。
从 NATS Go 客户端 v1.10+ 开始，引入了 global inbox subscription 优化：
连接建立时，客户端订阅一个共享前缀：
SUB _INBOX.AbCdEfGh.* 1
所有 Request 复用这个订阅，通过不同的后缀区分：
Request 1: _INBOX.AbCdEfGh.1
Request 2: _INBOX.AbCdEfGh.2
Request 3: _INBOX.AbCdEfGh.3这样整个连接生命周期内只需要一次 SUB，大幅减少协议开销。
3. 完整交互时序图
3.1 基础 Request/Reply 时序
Client App NATS Server Service App
| | |
|-- SUB _INBOX.abc.* (连接建立时自动) -->|
| | |
|-- PUB service.greet |
| ReplyTo: _INBOX.abc.1 ------------>|
| | |
| |-- MSG service.greet|
| | ReplyTo: _INBOX.abc.1
| |------------------>|
| | |
| | |-- 处理请求
| | |
| |<-- PUB _INBOX.abc.1|
| | payload: "Hello" |
| | |
|<-- MSG _INBOX.abc.1 (通过现有订阅) ---|
| payload: "Hello" |
| | |
|-- UNSUB _INBOX.abc.1 (可选清理) ---->|
| | |3.2 多节点集群中的路由时序
Client App NATS Node-1 NATS Node-2 Service App
| | | |
| | (Node-2 订阅了 service.greet) |
| |<--- RS+ service.greet ----------|
| | (路由传播：Node-2 有订阅者) |
| | | |
|-- PUB service.greet ---------->| |
| ReplyTo: _INBOX.xxx.1 | |
| | |
| Node-1 将消息路由给 Node-2 |
| |-- RMSG service.greet ---------->|
| | ReplyTo: _INBOX.xxx.1 |
| | |-- MSG -------->|
| | | |
| | | 处理请求
| | | |
| | |<-- PUB _INBOX.xxx.1
| | | |
| Node-2 将应答路由给 Node-1（Client连接处）
| |<-- RMSG _INBOX.xxx.1 -----------|
| | | |
|<-- MSG _INBOX.xxx.1 -----------| |
| payload: response | |注：RS+ 是 NATS 集群内部的 Route Subscribe 信号，详见第 08 篇集群文章。
4. 超时处理
4.1 为什么必须设置超时
在异步消息系统中，以下情况都会导致请求永远得不到响应：
- 没有任何服务实例订阅目标 subject
- 服务实例崩溃，来不及回复
- 网络分区导致消息丢失
- 服务处理时间超出预期
NATS Request 必须传入 timeout 参数，超时后返回 nats.ErrTimeout。
4.2 超时策略
go
// 方式 1：固定超时（简单场景）
msg, err := nc.Request("service.compute", data, 5*time.Second)
// 方式 2：基于上下文的超时（推荐，可集成 context 链）
ctx, cancel := context.WithTimeout(parentCtx, 5*time.Second)
defer cancel()
msg, err := nc.RequestWithContext(ctx, "service.compute", data)
// 方式 3：动态超时（根据 payload 大小估算）
timeout := baseTimeout + time.Duration(len(data)/1024)*time.Millisecond*100
msg, err := nc.Request("service.compute", data, timeout)4.3 超时时序
Client NATS Service
| | |
|-- Request(5s) ---->| |
| |-- MSG ------------->|
| | |
| | (处理中...)
| | |
|<-------- 5秒超时 --| |
| ErrTimeout | |
| | |
| | (5.1秒后才发布回复)
| |<-- PUB _INBOX.xxx --|
| | |
| (客户端已取消订阅，消息被丢弃) |
| | |注意：超时后客户端取消了对 inbox 的监听，服务端迟来的回复会被 NATS 丢弃（因为没有订阅者）。这是符合预期的行为，不会造成内存泄漏。
4.4 超时重试
go
func requestWithRetry(nc *nats.Conn, subject string, data []byte, timeout time.Duration, maxRetries int) (*nats.Msg, error) {
var lastErr error
for i := 0; i < maxRetries; i++ {
if i > 0 {
// 指数退避：100ms, 200ms, 400ms...
backoff := time.Duration(100<<uint(i-1)) * time.Millisecond
time.Sleep(backoff)
}
msg, err := nc.Request(subject, data, timeout)
if err == nil {
return msg, nil
}
lastErr = err
// 只有超时才重试，ErrNoResponders 重试无意义
if !errors.Is(err, nats.ErrTimeout) {
break
}
}
return nil, fmt.Errorf("after %d retries: %w", maxRetries, lastErr)
}5. 多应答模式
5.1 第一个回复胜出（默认行为）
nc.Request() 默认只等待第一个回复，收到后立即返回。这适合"最快响应"场景：
Service Instance A
Client --PUB------->
Service Instance B (Queue Group: 只有一个收到消息)
Service Instance C5.2 等待多个回复（Scatter-Gather 模式）
有时需要向多个服务广播查询，收集所有回复（例如：查询集群中所有节点的状态）。
使用 nc.PublishRequest() + 手动订阅实现：
go
// Scatter-Gather：向所有订阅者广播并收集回复
func scatterGather(nc *nats.Conn, subject string, data []byte, expectedReplies int, timeout time.Duration) ([]*nats.Msg, error) {
// 创建收件箱
inbox := nats.NewInbox()
var mu sync.Mutex
var replies []*nats.Msg
// 订阅收件箱
sub, err := nc.Subscribe(inbox, func(msg *nats.Msg) {
mu.Lock()
replies = append(replies, msg)
mu.Unlock()
})
if err != nil {
return nil, err
}
defer sub.Unsubscribe()
// 发布请求（注意：不使用 Queue Group，这样所有实例都能收到）
if err := nc.PublishRequest(subject, inbox, data); err != nil {
return nil, err
}
// 等待指定数量的回复或超时
deadline := time.Now().Add(timeout)
for {
mu.Lock()
count := len(replies)
mu.Unlock()
if count >= expectedReplies {
break
}
if time.Now().After(deadline) {
break
}
time.Sleep(10 * time.Millisecond)
}
mu.Lock()
defer mu.Unlock()
return replies, nil
}5.3 流式多应答（Streaming Response）
对于大文件传输、AI 流式输出等场景，服务端可以发送多个 chunk，客户端逐块接收：
go
// 服务端：流式回复
nc.Subscribe("service.stream", func(msg *nats.Msg) {
if msg.Reply == "" {
return // 无回信地址，忽略
}
chunks := []string{"chunk1", "chunk2", "chunk3"}
for _, chunk := range chunks {
nc.Publish(msg.Reply, []byte(chunk))
time.Sleep(100 * time.Millisecond) // 模拟生成延迟
}
// 发送结束标记
nc.Publish(msg.Reply, []byte("__DONE__"))
})
// 客户端：接收流式回复
inbox := nats.NewInbox()
sub, _ := nc.SubscribeSync(inbox)
defer sub.Unsubscribe()
nc.PublishRequest("service.stream", inbox, []byte("start"))
for {
msg, err := sub.NextMsg(5 * time.Second)
if err != nil {
break // 超时
}
if string(msg.Data) == "__DONE__" {
break
}
fmt.Printf("收到 chunk: %s\n", msg.Data)
}6. 服务端负载模式
6.1 单一服务实例
Client ──▶ NATS ──▶ Service (单实例)
│
单点故障风险
性能瓶颈这是最简单的模式，适合开发测试环境，生产不推荐。
6.2 多实例 + 普通订阅（广播）
┌──▶ Service Instance 1
Client ──▶ NATS──▶ Service Instance 2 (所有实例都收到！)
└──▶ Service Instance 3当多个实例订阅同一个 subject 时，NATS 默认将消息广播给所有订阅者。这导致一个请求被处理多次，通常不是期望行为。
6.3 多实例 + Queue Groups（推荐）
┌──▶ Service Instance 1 ┐
Client ──▶ NATS──▶ Service Instance 2 ├── Queue Group "svc"
└──▶ Service Instance 3 ┘
│
只有一个实例收到消息（随机选择）Queue Groups 确保同一 group 内只有一个成员处理每条消息，完美实现负载均衡。
7. 错误处理
7.1 ErrNoResponders
当没有任何订阅者监听目标 subject 时，NATS 服务端（v2.2+）会立即返回 ErrNoResponders，而不是等到超时。
Client NATS (无服务订阅)
| |
|-- Request -------->|
| |-- 检查：没有订阅者
| |
|<-- Status: 503 ----| (No Responders)
| ErrNoResponders |这是 NATS 2.2 引入的重要特性，避免客户端等待整个超时周期。
处理代码：
go
msg, err := nc.Request("service.greet", data, 5*time.Second)
if err != nil {
switch {
case errors.Is(err, nats.ErrNoResponders):
// 服务未启动 / subject 名称错误 / 服务全部下线
log.Println("没有可用的服务实例，请检查服务是否运行")
return ErrServiceUnavailable
case errors.Is(err, nats.ErrTimeout):
// 服务存在但响应超时（可能过载）
log.Println("服务响应超时，可考虑重试")
return ErrServiceTimeout
case errors.Is(err, nats.ErrConnectionClosed):
// NATS 连接已断开
log.Println("NATS 连接断开")
return ErrConnectionLost
default:
return fmt.Errorf("request failed: %w", err)
}
}7.2 服务端异常回复
建议服务端在遇到错误时，通过 Respond 发送结构化错误，而非静默失败：
go
type Response struct {
Success bool `json:"success"`
Data json.RawMessage `json:"data,omitempty"`
Error string `json:"error,omitempty"`
Code int `json:"code,omitempty"`
}
// 服务端统一错误回复
func respondError(msg *nats.Msg, code int, errMsg string) {
resp := Response{
Success: false,
Error: errMsg,
Code: code,
}
data, _ := json.Marshal(resp)
msg.Respond(data)
}8. Go 代码示例：完整 Request/Reply
8.1 服务端（Server）
go
package main
import (
"encoding/json"
"fmt"
"log"
"os"
"os/signal"
"syscall"
"time"
"github.com/nats-io/nats.go"
)
// 请求结构
type GreetRequest struct {
Name string `json:"name"`
}
// 响应结构
type GreetResponse struct {
Greeting string `json:"greeting"`
Timestamp time.Time `json:"timestamp"`
ServerID string `json:"server_id"`
}
func main() {
// 连接 NATS
nc, err := nats.Connect(nats.DefaultURL,
nats.Name("greet-service"),
nats.MaxReconnects(-1), // 无限重连
nats.ReconnectWait(2*time.Second),
)
if err != nil {
log.Fatalf("连接 NATS 失败: %v", err)
}
defer nc.Drain() // 优雅关闭：等待所有消息处理完成
serverID := fmt.Sprintf("server-%d", os.Getpid())
log.Printf("[%s] 启动服务，订阅 service.greet", serverID)
// 订阅请求 subject（使用 Queue Group 支持多实例）
sub, err := nc.QueueSubscribe("service.greet", "greet-workers", func(msg *nats.Msg) {
// 解析请求
var req GreetRequest
if err := json.Unmarshal(msg.Data, &req); err != nil {
log.Printf("解析请求失败: %v", err)
// 返回错误响应
if msg.Reply != "" {
msg.Respond([]byte(`{"error":"invalid request format"}`))
}
return
}
log.Printf("[%s] 收到请求：name=%s, reply=%s", serverID, req.Name, msg.Reply)
// 模拟处理
time.Sleep(10 * time.Millisecond)
// 构建响应
resp := GreetResponse{
Greeting: fmt.Sprintf("Hello, %s! 👋", req.Name),
Timestamp: time.Now(),
ServerID: serverID,
}
data, err := json.Marshal(resp)
if err != nil {
log.Printf("序列化响应失败: %v", err)
return
}
// 发送响应到 reply-to subject
if err := msg.Respond(data); err != nil {
log.Printf("发送响应失败: %v", err)
}
})
if err != nil {
log.Fatalf("订阅失败: %v", err)
}
defer sub.Drain()
// 等待退出信号
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Printf("[%s] 正在优雅关闭...", serverID)
}8.2 客户端（Client）
go
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"time"
"github.com/nats-io/nats.go"
)
type GreetRequest struct {
Name string `json:"name"`
}
type GreetResponse struct {
Greeting string `json:"greeting"`
Timestamp time.Time `json:"timestamp"`
ServerID string `json:"server_id"`
}
// GreetClient 封装 Request/Reply 调用
type GreetClient struct {
nc *nats.Conn
}
func NewGreetClient(nc *nats.Conn) *GreetClient {
return &GreetClient{nc: nc}
}
// Greet 发送问候请求（带 context 超时控制）
func (c *GreetClient) Greet(ctx context.Context, name string) (*GreetResponse, error) {
// 序列化请求
req := GreetRequest{Name: name}
data, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("序列化请求失败: %w", err)
}
// 发送请求（使用 context 控制超时）
msg, err := c.nc.RequestWithContext(ctx, "service.greet", data)
if err != nil {
switch {
case err == nats.ErrNoResponders:
return nil, fmt.Errorf("greet 服务不可用（无订阅者）")
case err == context.DeadlineExceeded || err == nats.ErrTimeout:
return nil, fmt.Errorf("greet 服务响应超时")
default:
return nil, fmt.Errorf("请求失败: %w", err)
}
}
// 解析响应
var resp GreetResponse
if err := json.Unmarshal(msg.Data, &resp); err != nil {
return nil, fmt.Errorf("解析响应失败: %w", err)
}
return &resp, nil
}
func main() {
nc, err := nats.Connect(nats.DefaultURL)
if err != nil {
log.Fatalf("连接 NATS 失败: %v", err)
}
defer nc.Close()
client := NewGreetClient(nc)
// 示例 1：正常请求
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
resp, err := client.Greet(ctx, "NATS")
if err != nil {
log.Printf("请求失败: %v", err)
} else {
fmt.Printf("响应：%s（来自 %s，时间：%v）\n",
resp.Greeting, resp.ServerID, resp.Timestamp.Format(time.RFC3339))
}
// 示例 2：并发请求（展示不同实例处理）
fmt.Println("\n--- 并发请求测试 ---")
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
resp, err := client.Greet(ctx, fmt.Sprintf("User-%d", i))
if err != nil {
fmt.Printf("[%d] 错误: %v\n", i, err)
return
}
fmt.Printf("[%d] %s -> 由 %s 处理\n", i, resp.Greeting, resp.ServerID)
}(i)
}
wg.Wait()
}9. Queue Groups 概念
9.1 什么是 Queue Group
Queue Group（队列组）是 NATS 的一种订阅模式，它允许多个订阅者共享同一个 subject 的消息负载：
- 普通订阅：消息广播给所有订阅者
- Queue 订阅：消息只发给 group 内的某一个订阅者
Queue Group 就是 NATS 内置的负载均衡器，不需要额外的代理或中间件。
9.2 关键特性
1. 同一 Queue Group 内，每条消息只有一个成员收到
2. 不同 Queue Group 之间，互相独立（都会收到消息）
3. Queue Group 成员可动态加入/离开（无需重启）
4. 故障成员自动剔除（断开连接即移除）
5. 无需在服务端配置，纯客户端行为10. Queue Groups vs 普通订阅
10.1 对比图
场景：3 个实例都订阅 "orders.process"，发布 1 条消息
普通订阅（Fan-out）：
┌─────────────────────────────────────────────┐
│ │
│ Publisher │
│ │ │
│ └──PUB "orders.process"──▶ NATS │
│ │ │
│ ┌───────┼───────┐ │
│ ▼ ▼ ▼ │
│ Sub-1 Sub-2 Sub-3 │
│ (收到) (收到) (收到) │
│ 消息被处理了 3 次！ │
└─────────────────────────────────────────────┘
Queue Group 订阅（Load Balance）：
┌─────────────────────────────────────────────┐
│ │
│ Publisher │
│ │ │
│ └──PUB "orders.process"──▶ NATS │
│ │ │
│ 随机选1 │
│ ┌───────┤───────┐ │
│ ▼ │ │ │
│ Sub-1 Sub-2 Sub-3 │
│ (收到) (未收到)(未收到)│
│ 消息只被处理了 1 次！ │
└─────────────────────────────────────────────┘10.2 混合模式（Queue Group + 普通订阅并存）
场景：监控系统 + 处理系统 并存
Publisher
│
└──PUB "orders.process"──▶ NATS
│
┌───────────┼────────────────┐
▼ ▼ ▼
Monitor-Sub Worker-1 ┐ Worker-2 ┐
(普通订阅, Worker-2 ├─ Queue Worker-3 ├─ Queue
每条都收到) Worker-3 ┘ "proc" Worker-4 ┘ "audit"
- Monitor 每条消息都记录日志
- "proc" group 中 1 个 worker 处理业务
- "audit" group 中 1 个 worker 做审计11. 分发策略
11.1 NATS 的随机分发
NATS 服务端对 Queue Group 的分发策略是随机选择（不是严格轮询）：
消息序列：M1 M2 M3 M4 M5 M6
Workers: W1 W2 W3
可能的分发结果（随机）：
M1 → W2
M2 → W2 (连续两次给同一个)
M3 → W1
M4 → W3
M5 → W1
M6 → W3这种随机策略的优点：
- 实现简单，服务端无需维护轮询状态
- 无状态，worker 动态增减无需调整
- 统计上均匀，大量消息下各 worker 负载趋于相等
对于需要严格顺序处理的场景，应使用 JetStream Consumer（见第 14 节）。
11.2 连接数与分发粒度
重要：Queue Group 的分发单位是订阅（subscription），不是连接（connection）。
如果一个进程内对同一 Queue Group 创建了多个订阅，则它会被选中多次（权重更高）：
go
// 不推荐！同一进程内多次 QueueSubscribe 同一 group
// 会导致该进程获得比其他进程更多的消息
sub1, _ := nc.QueueSubscribe("orders", "workers", handler) // 1/4 概率
sub2, _ := nc.QueueSubscribe("orders", "workers", handler) // 1/4 概率
// 另外 2 个进程各 1 个订阅，各 1/4 概率12. Queue Groups 实际应用场景
12.1 水平扩展服务（最常见）
┌─── OrderService Pod-1 ─── QueueSub("orders", "svc")
Client ─▶ NATS ────├─── OrderService Pod-2 ─── QueueSub("orders", "svc")
└─── OrderService Pod-3 ─── QueueSub("orders", "svc")
优点：
- Pod 数量可随时增减（K8s HPA）
- 无需修改 Client 代码
- 无需 LoadBalancer
- Pod 重启不影响其他实例实际部署时，只需在 Deployment 中 replicas: 3，每个实例启动时调用 QueueSubscribe，即可自动组成 Queue Group。
12.2 工作队列（Task Queue）
任务生产者
│
└──PUB "tasks.image.resize"──▶ NATS
│
Queue Group "image-workers"
┌─────────┼─────────┐
▼ ▼ ▼
Worker1 Worker2 Worker3
（各自独立处理，互不干扰）
特点：
- 任务确保被处理一次（无重复）
- Worker 崩溃后任务"丢失"（需 JetStream 持久化）
- 适合幂等任务（纯 Core NATS）12.3 热备份（Active-Passive Failover）
正常状态：
Publisher ──▶ NATS ──▶ Primary Worker（处理）
└── Standby Worker（等待，不处理）
等等，这不对！Queue Group 是随机分发，无法保证始终发给 Primary。
正确实现热备份应使用 JetStream 的 Pull Consumer + AckWait，
或者使用 leader election 机制（见注意事项）。注意：Queue Group 不适合实现严格的 Active-Passive 热备份，因为 NATS 无法保证消息始终路由到特定成员。如需热备份，应使用 JetStream Consumer 的 deliver_policy 结合 ack_wait，让消息在未确认时自动重发给其他消费者。
12.4 地理分区处理
全球用户请求 ──▶ NATS Cluster
│
┌─────────────┼──────────────┐
▼ ▼ ▼
Asia Workers EU Workers US Workers
QueueSub("req", QueueSub("req", QueueSub("req",
"asia-svc") "eu-svc") "us-svc")
不同 Queue Group 各自独立接收所有消息，
实现按地区复制处理（审计、统计等）13. Go 代码示例：Queue Groups Worker Pool
13.1 启动 3 个 Worker，Queue Group 分发
go
package main
import (
"encoding/json"
"fmt"
"log"
"os"
"os/signal"
"sync"
"syscall"
"time"
"github.com/nats-io/nats.go"
)
type Task struct {
ID string `json:"id"`
Type string `json:"type"`
Payload json.RawMessage `json:"payload"`
}
type TaskResult struct {
TaskID string `json:"task_id"`
WorkerID string `json:"worker_id"`
Success bool `json:"success"`
Result string `json:"result,omitempty"`
Error string `json:"error,omitempty"`
Duration time.Duration `json:"duration_ms"`
}
// Worker 封装
type Worker struct {
id string
nc *nats.Conn
sub *nats.Subscription
stats struct {
processed uint64
errors uint64
}
}
func NewWorker(id string, nc *nats.Conn) (*Worker, error) {
w := &Worker{
id: id,
nc: nc,
}
// 加入 Queue Group "task-workers"
sub, err := nc.QueueSubscribe("tasks.process", "task-workers", w.handleTask)
if err != nil {
return nil, fmt.Errorf("worker %s 订阅失败: %w", id, err)
}
w.sub = sub
log.Printf("[Worker %s] 已启动，加入 Queue Group 'task-workers'", id)
return w, nil
}
func (w *Worker) handleTask(msg *nats.Msg) {
start := time.Now()
var task Task
if err := json.Unmarshal(msg.Data, &task); err != nil {
log.Printf("[Worker %s] 解析任务失败: %v", w.id, err)
w.stats.errors++
if msg.Reply != "" {
result := TaskResult{
WorkerID: w.id,
Success: false,
Error: "invalid task format",
}
data, _ := json.Marshal(result)
msg.Respond(data)
}
return
}
log.Printf("[Worker %s] 开始处理任务 %s (type=%s)", w.id, task.ID, task.Type)
// 模拟不同类型任务的处理时间
var processingTime time.Duration
switch task.Type {
case "fast":
processingTime = 10 * time.Millisecond
case "medium":
processingTime = 100 * time.Millisecond
case "slow":
processingTime = 500 * time.Millisecond
default:
processingTime = 50 * time.Millisecond
}
time.Sleep(processingTime)
// 构建结果
result := TaskResult{
TaskID: task.ID,
WorkerID: w.id,
Success: true,
Result: fmt.Sprintf("任务 %s 处理完成（耗时 %v）", task.ID, processingTime),
Duration: time.Since(start),
}
w.stats.processed++
// 如果有 reply subject，发送结果（支持 Request/Reply 模式）
if msg.Reply != "" {
data, _ := json.Marshal(result)
if err := msg.Respond(data); err != nil {
log.Printf("[Worker %s] 发送结果失败: %v", w.id, err)
}
}
log.Printf("[Worker %s] 任务 %s 完成，总处理: %d", w.id, task.ID, w.stats.processed)
}
func (w *Worker) Drain() {
w.sub.Drain()
log.Printf("[Worker %s] 已停止，共处理任务: %d，错误: %d",
w.id, w.stats.processed, w.stats.errors)
}
// 任务发布者（模拟客户端）
func publishTasks(nc *nats.Conn, count int) {
log.Printf("开始发布 %d 个任务...", count)
taskTypes := []string{"fast", "medium", "slow"}
var wg sync.WaitGroup
for i := 1; i <= count; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
task := Task{
ID: fmt.Sprintf("task-%04d", i),
Type: taskTypes[i%len(taskTypes)],
}
data, _ := json.Marshal(task)
// 使用 Request 获取处理结果
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
msg, err := nc.RequestWithContext(ctx, "tasks.process", data)
if err != nil {
log.Printf("任务 %s 发送失败: %v", task.ID, err)
return
}
var result TaskResult
if err := json.Unmarshal(msg.Data, &result); err != nil {
log.Printf("任务 %s 结果解析失败: %v", task.ID, err)
return
}
log.Printf("任务 %s 由 Worker %s 处理完成（%v）",
result.TaskID, result.WorkerID, result.Duration)
}(i)
// 稍微错开发送时间，避免所有任务同时到达
time.Sleep(20 * time.Millisecond)
}
wg.Wait()
log.Println("所有任务已完成")
}
func main() {
nc, err := nats.Connect(nats.DefaultURL,
nats.MaxReconnects(-1),
nats.ReconnectWait(2*time.Second),
)
if err != nil {
log.Fatalf("连接 NATS 失败: %v", err)
}
defer nc.Close()
// 启动 3 个 Worker
workers := make([]*Worker, 3)
for i := 0; i < 3; i++ {
w, err := NewWorker(fmt.Sprintf("W%d", i+1), nc)
if err != nil {
log.Fatalf("启动 Worker %d 失败: %v", i+1, err)
}
workers[i] = w
}
// 等待 Worker 准备就绪
time.Sleep(100 * time.Millisecond)
// 发布任务（在单独 goroutine 中运行）
go publishTasks(nc, 15)
// 等待退出信号
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Println("正在优雅关闭所有 Worker...")
for _, w := range workers {
w.Drain()
}
}13.2 预期输出（展示负载分布）
[Worker W1] 已启动，加入 Queue Group 'task-workers'
[Worker W2] 已启动，加入 Queue Group 'task-workers'
[Worker W3] 已启动，加入 Queue Group 'task-workers'
开始发布 15 个任务...
[Worker W2] 开始处理任务 task-0001 (type=fast)
[Worker W1] 开始处理任务 task-0002 (type=medium)
[Worker W3] 开始处理任务 task-0003 (type=slow)
[Worker W2] 开始处理任务 task-0004 (type=fast)
[Worker W2] 任务 task-0001 完成，总处理: 1
[Worker W1] 开始处理任务 task-0005 (type=medium)
...
任务分布（随机，最终趋于均匀）：
W1: 5 个任务
W2: 6 个任务
W3: 4 个任务14. Queue Subscribe vs JetStream Consumer
14.1 核心区别
Queue Subscribe (Core NATS)：
发布 ──▶ NATS ──▶ 立即路由 ──▶ 一个 Worker 处理
│
如果没有订阅者
│
消息丢失！
JetStream Consumer (Pull/Push)：
发布 ──▶ NATS Stream（持久化存储）
│
Worker 来拉取（Pull）或 Push 投递
│
Worker 处理后 Ack
│
未 Ack 的消息自动重投14.2 选择指南
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 实时 RPC，低延迟 | Queue Subscribe | 无存储开销，延迟极低 |
| 任务不能丢失 | JetStream Pull Consumer | 持久化 + Ack 保证 |
| Worker 可能下线 | JetStream Pull Consumer | 消息等待 Worker 重连 |
| 消息需要重放 | JetStream | 支持 seek_by_time 等 |
| 简单负载均衡 | Queue Subscribe | 无需配置 Stream |
| 精确 Exactly-Once | JetStream + Dedup | 消息去重支持 |
| 高吞吐实时处理 | Queue Subscribe | 省去 Ack 往返 |
| 跨服务审计追踪 | JetStream | 消息可重放查询 |
14.3 混用场景
go
// 实时命令用 Queue Subscribe（低延迟）
nc.QueueSubscribe("cmd.reboot", "cmd-workers", handleReboot)
// 重要订单用 JetStream Consumer（不能丢）
js, _ := nc.JetStream()
js.Subscribe("orders.new", handleOrder,
nats.Durable("order-processor"),
nats.DeliverNew(),
nats.AckExplicit(),
nats.Queue("order-workers"), // JetStream 的 Queue Group
)15. 实际架构示例：API Gateway + Queue Group + Service
15.1 架构图
┌─────────────────────────────────────┐
│ Kubernetes Cluster │
│ │
HTTP Client │ ┌──────────────┐ │
│ │ │ API Gateway │ │
│──POST /execute──▶│ │ (1 Pod) │ │
│◀──SSE stream── │ │ │ │
│ │ nc.Request()│ │
│ │ "svc.exec" │ │
│ └──────┬───────┘ │
│ │ NATS Request │
│ ▼ │
│ ┌─────────────────────────────┐ │
│ │ NATS Cluster │ │
│ │ (3-node, StatefulSet) │ │
│ └──────────────────────────────┘ │
│ │ Route to Queue Group │
│ ▼ │
│ ┌──────────────────────────────┐ │
│ │ Exec Service │ │
│ │ ┌────────┐ ┌────────┐ │ │
│ │ │ Pod-1 │ │ Pod-2 │ ... │ │
│ │ │QSub │ │QSub │ │ │
│ │ │"svc. │ │"svc. │ │ │
│ │ │exec" │ │exec" │ │ │
│ │ │"exec- │ │"exec- │ │ │
│ │ │workers"│ │workers"│ │ │
│ │ └────────┘ └────────┘ │ │
│ │ HPA: 2~10 replicas │ │
│ └──────────────────────────────┘ │
└─────────────────────────────────────┘15.2 API Gateway 代码
go
// API Gateway：HTTP 请求 → NATS Request/Reply
func (h *GatewayHandler) Execute(w http.ResponseWriter, r *http.Request) {
var req ExecuteRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "invalid request", http.StatusBadRequest)
return
}
// 设置 SSE 响应头
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming not supported", http.StatusInternalServerError)
return
}
// 通过 NATS 发送到 Exec Service（Queue Group 自动负载均衡）
ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
defer cancel()
natsReq, _ := json.Marshal(req)
msg, err := h.nc.RequestWithContext(ctx, "svc.exec", natsReq)
if err != nil {
if err == nats.ErrNoResponders {
http.Error(w, "exec service unavailable", http.StatusServiceUnavailable)
} else {
http.Error(w, "request timeout", http.StatusGatewayTimeout)
}
return
}
// 将响应转为 SSE 事件
fmt.Fprintf(w, "data: %s\n\n", msg.Data)
flusher.Flush()
}15.3 Exec Service 代码（多实例）
go
// Exec Service：每个 Pod 启动时加入 Queue Group
func startExecService(nc *nats.Conn, podName string) {
nc.QueueSubscribe("svc.exec", "exec-workers", func(msg *nats.Msg) {
var req ExecuteRequest
json.Unmarshal(msg.Data, &req)
log.Printf("[%s] 执行命令：device=%s", podName, req.DeviceID)
result := executeOnDevice(req) // 实际执行逻辑
resp, _ := json.Marshal(result)
msg.Respond(resp)
})
}16. 小结
Request/Reply 要点回顾
1. _INBOX 是自动生成的临时 reply-to subject
2. 客户端用全局共享 inbox 前缀（性能优化）
3. 必须设置超时，推荐使用 context.WithTimeout
4. ErrNoResponders 表示无服务在线（立即失败，不等超时）
5. 流式应答通过多次 Publish 到 reply subject 实现Queue Groups 要点回顾
1. 同组内只有一个成员收到每条消息
2. 不同组之间互相独立（各自收到所有消息）
3. 成员动态增减，NATS 自动调整
4. 分发策略为随机，统计上均匀
5. 不适合严格顺序 / 不丢失场景（用 JetStream）
6. 与 Request/Reply 结合实现可扩展 RPC选择矩阵
┌─────────────────────┬──────────────────┬──────────────────┐
│ 需求 │ 推荐模式 │ 备注 │
├─────────────────────┼──────────────────┼──────────────────┤
│ 同步 RPC 调用 │ Request/Reply │ 带超时 │
│ 广播查询（所有实例）│ Scatter-Gather │ 无 Queue Group │
│ 负载均衡处理 │ Queue Group │ 随机分发 │
│ 任务不能丢失 │ JetStream Consumer│ 持久化 + Ack │
│ 流式响应 │ Request + 多Pub │ 需要协议约定 │
│ 健康检查/服务发现 │ Request/Reply │ ErrNoResponders │
└─────────────────────┴──────────────────┴──────────────────┘