Go语言实战：NATS JetStream消息持久化与消费模式详解

1. 内容整体设计与思路拆解

1.1 从“消息队列”到“JetStream”：为什么不用原生 NATS？

先交代背景。很多人一听“NATS”，第一反应是“那个轻量级消息中间件”，然后默认它和老牌 MQ（RabbitMQ、Kafka）一样，把消息持久化、重放、消费者确认这些功能都内置好了。实际上原生 NATS 核心是一个纯实时发布订阅系统，消息发出去以后，如果当时没有订阅者在线，或者网络抖动导致客户端没收到，那这条消息就这么丢了，不会有任何“补偿”机制。

NATS 官方也清楚这个短板，所以推出了 JetStream。它是在 NATS 基础上构建的存储与流式处理层，本质上就是把消息“落盘”，提供持久化、至少一次投递、消费者分组、消息重放等功能。简单说，原生 NATS 像“对讲机”，喊完就完事；JetStream 像“录音电话”，你没接到，回头还能回放。

这个项目标题是“golang如何使用NATS JetStream”，关键词里只写了 golang、NATS、JetStream，但真正要解决的是“在 Go 项目里怎么把 JetStream 用起来、用得稳”。所以在设计博文内容时，我没有单纯把官方文档翻译一遍，而是按照“环境准备 → 核心概念 → 生产消费 → 持久化与重放 → 运维排查”这条主线来拆，目标是让读者看完就能在自己项目里落一套可用的 JetStream 基础链路。

1.2 方案选型：为什么不选 Kafka 或 RabbitMQ？

在实际项目里，选消息中间件经常会纠结。如果是大数据量、需要复杂分区和顺序保证的场景，Kafka 确实是默认选择；如果需求是灵活路由、多协议支持，RabbitMQ 也很成熟。但 JetStream 的定位是“云原生、轻量、运维简单”，它不需要像 Kafka 那样依赖 Zookeeper（新版本虽然去掉了 ZK，但整体运维复杂度还是比 NATS 高），也不像 RabbitMQ 那样需要单独管理 Exchange、Queue、Binding 一堆概念。

JetStream 在很多场景下是“够用且更省心”的方案，尤其是微服务之间的异步解耦、事件驱动、边缘计算这类对资源占用敏感的场景。Golang 作为 NATS 官方的一等公民语言，客户端库非常完善，这两者搭配起来非常顺手。

1.3 我能提供什么信息

这篇文章主要是写给两类人：一类是从没接触过 JetStream 的 Go 开发者，想快速跑通一条链路；另一类是已经用过原生 NATS，但被“消息丢失”“没有持久化”坑过，想了解 JetStream 迁移方案的人。我会从“核心概念”开始讲，再把生产、消费、持久化、重放这些关键环节拆开，每一步都有可运行的 Go 代码和踩坑记录。文末还会整理一份常见问题速查表，基本覆盖我在本地和测试环境里被折腾过的问题。

2. 核心概念速览：Stream、Consumer、Message

2.1 Stream：消息的“存储桶”

JetStream 里最核心的概念就是 Stream。你可以把它理解成一个“消息存储桶”或“日志文件”。生产者往 Stream 里发消息，Stream 负责把消息持久化到磁盘（也可以配成内存模式），并按照一定规则保留一段时间或一定数量。Stream 还支持按 subject 过滤，也就是说一个 Stream 可以承载多个 subject 的消息。

创建 Stream 时最关键的几个参数是：

• Name（stream 名称）：唯一标识；
• Subjects（主题列表）：这个 Stream 订阅哪些 subject；
• Storage（存储类型）：file（落盘）或 memory（纯内存）；
• Retention（保留策略）：LimitsPolicy（按数量/时间/大小限制）、InterestPolicy（所有消费者确认后删除）、WorkQueuePolicy（工作队列模式）。

Stream 命名是有讲究的，建议用“业务线-环境-用途”这种格式，比如 order-prod-event、user-dev-task。项目里如果直接叫 test、demo，后面管理起来会很混乱。

2.2 Consumer：消息的“读取游标”

Consumer 是绑定在某个 Stream 上的“读取器”。它记录了自己消费到了哪条消息，支持单独持久化消费进度。Consumer 分两种：

• Push Consumer：JetStream 把消息主动推送给客户端；
• Pull Consumer：客户端主动拉取一批消息。

这两种模式的适用场景不同。Push Consumer 适合消息量可控、客户端处理能力稳定、需要低延迟的场景，但客户端必须保证在线，否则消息会积压或者触发流控。Pull Consumer 适合一批一批处理、客户端可以随时上线离线的场景（比如定时任务），也适合多个消费者实例横向扩展分摊消息。

从个人体验来说，如果只是“接消息处理一下”，Push 模式写起来更直观；但如果要处理大批量、需要精细控制消费速率，Pull 模式更稳。官方也更加推荐 Pull Consumer，因为更容易做负载均衡和按需拉取。

2.3 Message、Subject、Queue Group

Message 就是实际传递的数据，除了业务 payload，还自带一个头部集合、发布时间、重放次数等元数据。Subject 是消息的路由地址，发布者往某个 subject 发消息，Stream 根据配置决定把哪些 subject 的消息收进“桶里”。

Queue Group 是 NATS 里做负载均衡的机制。同一个队列组里的多个订阅者，一条消息只会被其中一个消费。JetStream 的 Consumer 天然支持队列模式，创建 Consumer 时可以指定队列名称，这样多个客户端实例共享这个 Consumer 时就自动分摊消息。

3. 环境准备：本地跑通 NATS JetStream

3.1 下载安装 NATS Server

这一步很简单，直接去 NATS 官方 GitHub Releases 页面下载对应平台的二进制，Windows 选 nats-server-xxx-amd64.exe，Linux 选 nats-server-xxx-amd64.deb 或 tar.gz。也可以直接用 Docker：

docker run -d --name nats-server \ -p 4222:4222 -p 8222:8222 \ nats:latest

这里我把 4222（客户端端口）和 8222（监控端口）都映射出来了。8222 是 HTTP 监控接口，可以查看服务状态、连接数、Stream 统计，虽然生产环境一般不会直接暴露，但本地调试非常有用。

启动后在浏览器打开 http://localhost:8222 ，能看到 NATS 服务器的基础信息，包括版本、连接数、路由数、JetStream 是否启用。JetStream 默认是启用的，但也可以通过启动参数显式开启：

nats-server -js

3.2 验证 JetStream 状态

直接用官方命令行工具 nats 可以快速检查：

nats server report jetstream

如果输出显示 JetStream 状态为 Active，说明一切正常。看到这里就可以开始写 Go 代码了。

3.3 Go 客户端依赖安装

官方 Go 客户端库是 github.com/nats-io/nats.go，目前版本（以 v1.36.0 为例）支持 JetStream 所有核心功能。安装：

go get github.com/nats-io/nats.go

需要注意，nats.go 的 JetStream API 在 v1.28 之后有一些变化，部分旧示例代码跑不通，主要是 JetStreamManager 和 JetStreamContext 的创建方式、Pull Consumer 的 Fetch 方法签名等。如果参考老文章报错了，优先看官方仓库里的最新示例。

4. 用 Go 创建连接并初始化 JetStream 上下文

4.1 基础连接代码

创建一个 nats.go 文件，先把连接写好：

package main import ( "log" "time" "github.com/nats-io/nats.go" ) func main() { nc, err := nats.Connect( "nats://127.0.0.1:4222", nats.Name("go-jetstream-demo"), nats.Timeout(5*time.Second), nats.MaxReconnects(10), nats.ReconnectWait(2*time.Second), ) if err != nil { log.Fatalf("连接 NATS 失败: %v", err) } defer nc.Close() js, err := nc.JetStream() if err != nil { log.Fatalf("初始化 JetStream 上下文失败: %v", err) } log.Println("连接成功，JetStream 上下文已创建") }

nats.Connect 里几个参数值得解释一下：

• nats.Name：设置客户端名称，会显示在监控面板上，便于定位连接是哪台机器/哪个服务；
• nats.Timeout：连接超时时间，防止服务不可用时客户端无限卡住；
• nats.MaxReconnects：断线自动重连次数，设为 -1 表示无限重连，但生产环境建议设一个合理值，避免服务异常时客户端默默重试导致问题难以排查；
• nats.ReconnectWait：重连间隔。

4.2 JetStream 上下文的两种创建方式

nats.Connect 之后，可以通过 nc.JetStream() 拿到一个 JetStreamContext，它既能管理 Stream/Consumer，也能发布和订阅消息。另有一个 nc.JetStreamManager()，专门用于创建和删除 Stream、Consumer 等管理操作。后者返回的接口更偏管理侧，没有订阅发布方法。

项目里如果希望同时发布消息和管理资源，就直接用 JetStreamContext；如果只想做管理工具（比如一个控制台后台），用 JetStreamManager 更清晰。

4.3 连接池和生命周期建议

很多初写 JetStream 的 Go 代码会犯一个错误：每次发布消息都新建连接。NATS 客户端本身就支持连接复用，内部是异步消息处理，没有必要频繁开关连接。建议在进程启动时建立一个全局连接，最后在进程退出时 Close。

另外要注意 JetStream 的连接和普通 NATS 连接是同一个 TCP 连接，共用同一个连接对象即可，不需要额外开启别的端口。

5. 创建 Stream：把消息“存下来”

5.1 从代码里创建 Stream

一般建议在初始化阶段主动创建 Stream，而不是依赖运维手动去控制台创建。这样代码“自举”能力强，部署新环境时不用额外操作。

下面的代码实现“如果 Stream 不存在则创建”的逻辑：

func ensureStream(js nats.JetStreamContext, streamName, subject string) error { info, err := js.StreamInfo(streamName) if err != nil { if err != nats.ErrStreamNotFound { return err } _, err = js.AddStream(&nats.StreamConfig{ Name: streamName, Subjects: []string{subject}, Storage: nats.FileStorage, Retention: nats.LimitsPolicy, MaxAge: time.Hour * 24 * 7, MaxBytes: 1024 * 1024 * 1024, }) if err != nil { return err } return nil } _ = info return nil }

这里有几个参数值得细讲：

• Storage: nats.FileStorage，表示消息落盘。如果追求极低延迟且不关心崩溃丢消息，可以选 nats.MemoryStorage；
• Retention: nats.LimitsPolicy，表示按限制条件（时间/大小/数量）自动过期删除。如果希望“所有消费者都确认后删除”，用 InterestPolicy；如果希望每个消息只被处理一次（工作队列），用 WorkQueuePolicy；
• MaxAge: 7天，7天前的消息自动清理；
• MaxBytes: 1GB，总消息大小超限后从最旧的消息开始清理。

5.2 Stream 参数如何定：别拍脑袋

选参数要结合业务场景，比如用户行为日志，可能要保留 30 天；验证码短信这种消息，可能几分钟后就没用了，保留 1 小时即可。MaxAge 太长会造成磁盘占用持续增长，太短则可能导致消费端短暂故障时消息已经被淘汰，无法追溯。

计算磁盘占用可以简单估算：假设平均每条消息 2KB，高峰期每秒 1000 条，一天消息量是 1000 × 2KB × 86400 ≈ 172GB。这个量级用 MaxBytes 限制就很有必要，否则很容易把磁盘写爆。实际项目里要把消息压缩、副本备份这些也考虑进去。

5.3 一个坑：Subject 绑定后 Stream 不会自动修改

Stream 创建之后，如果后续需要增加新的 subject，很多人以为加在代码配置里、重启服务就能自动生效，其实不会。AddStream 只能创建时指定 Subjects，UpdateStream 才能更新。如果你直接修改代码里的 Subjects 再重启，StreamInfo 判断“已经存在”就直接跳过了，新 subject 不会进入这个 Stream。

如果确实需要扩展，有两种方式：一是手动用 UpdateStream / Stream.Update() 把新的 subject 加进配置；二是干脆创建新的 Stream 并指定新的 subject。前者会保留已有消息和历史消费进度，后者是隔离的，看业务需要。

6. 发布消息到 JetStream

6.1 基础发布代码

发布消息非常简单，关键在于“同步”还是“异步”。

同步发布：

func publishSync(js nats.JetStreamContext, subject, payload string) error { ack, err := js.Publish(subject, []byte(payload)) if err != nil { return err } log.Printf("发布成功, Stream=%s Sequence=%d", ack.Stream, ack.Sequence) return nil }

异步发布：

func publishAsync(js nats.JetStreamContext, subject, payload string) { pubAckFuture, err := js.PublishAsync(subject, []byte(payload)) if err != nil { log.Printf("异步发布失败: %v", err) return } select { case pubAck := <-pubAckFuture.Ok(): log.Printf("异步发布成功, Stream=%s Sequence=%d", pubAck.Stream, pubAck.Sequence) case err := <-pubAckFuture.Err(): log.Printf("异步发布错误: %v", err) case <-time.After(5 * time.Second): log.Printf("异步发布超时") } }

同步发布适合对可靠性要求高的场景，特别是刚接入 JetStream 时，强烈建议先同步发布，确认 Ack 正常后再切异步，便于定位问题。异步发布吞吐更高，但程序退出前需要调用 js.PublishAsyncComplete() 等待缓冲区的消息全部确认完成，否则可能丢消息。

6.2 Ack 是什么

JetStream 发布后服务器返回的 PubAck 对象里包含 Stream 名称和消息序号（Sequence）。消息序号是单调递增的，相当于这条消息在该 Stream 中的位置，可以用于重放、验证、排查。如果 Publish 返回 error，通常说明消息没有被存储成功，需要重试。

6.3 发布时给消息加 Header

JetStream 消息支持 Header，类似 HTTP Header。可以用 nats.Header 在发布时附加业务信息，比如消息来源、消息类型、版本号，消费端读取时无需解析 payload 就能拿到这些元数据。

hdr := make(nats.Header) hdr.Set("X-Source", "order-service") hdr.Set("X-Msg-Type", "order.created") ack, err := js.Publish(subject, []byte(payload), nats.Headers(hdr))

这个特性在多人协作、链路追踪时非常有用，但是很多示例代码不会提。具体业务里可以用 Header 传递 trace_id、app_id、环境标识等等，配合日志打好链路跟踪基础。

7. 消费消息：Push 和 Pull 两种姿势

7.1 Push Consumer 示例

先创建一个 Push Consumer，然后直接订阅。

func consumeWithPush(js nats.JetStreamContext, streamName, consumerName string) { _, err := js.AddConsumer(streamName, &nats.ConsumerConfig{ Durable: consumerName, AckPolicy: nats.AckExplicitPolicy, DeliverSubject: "order-push-deliver", }) if err != nil { log.Printf("创建 Consumer 失败: %v", err) return } sub, err := js.PullSubscribe("order.created", consumerName) if err != nil { log.Printf("Push 订阅失败: %v", err) return } for { select {} } }

Push 模式的订阅需要指定一个 DeliverSubject（投递主题），服务端会把消息推到该 Subject 上，客户端在后台自动接收。

这段代码其实有点“伪 Push”的意思，因为 PullSubscribe 仍然会创建一个消费循环。更直接的 Push 方式是用 Subscribe 配合 BindStream，不过整体概念上都是服务端推动、客户端回调。

为了不让示例复杂化，下面统一用一个更清晰的消费循环方式来说明。

7.2 Pull Consumer 示例（推荐）

Pull Consumer 更适合大多数项目，因为它可控性更强，消费者处理一批、拉取一批，天然适合横向扩展。

func consumeWithPull(js nats.JetStreamContext, streamName, consumerName string) { _, err := js.AddConsumer(streamName, &nats.ConsumerConfig{ Durable: consumerName, AckPolicy: nats.AckExplicitPolicy, }) if err != nil { log.Printf("创建 Consumer 失败: %v", err) return } sub, err := js.PullSubscribe("order.created", consumerName) if err != nil { log.Printf("Pull 订阅失败: %v", err) return } for { msgs, err := sub.Fetch(10, nats.MaxWait(5*time.Second)) if err != nil { if err == nats.ErrTimeout { continue } log.Printf("Fetch 错误: %v", err) time.Sleep(time.Second) continue } for _, msg := range msgs { log.Printf("收到消息: %s, 序列号: %v", string(msg.Data), msg.Metadata().Sequence) msg.Ack() } } }

这里有几个注意点：

• nats.AckExplicitPolicy：每条消息必须显式 Ack，否则等 AckWait 超时后消息会被重新投递；
• sub.Fetch(10) 表示一次最多拉 10 条，但实际拉取的数量可能小于 10，取决于当前积压消息数；
• nats.MaxWait(5s)：等待消息的最大时间，超时返回 ErrTimeout 但要继续循环，不能当作 Fatal；
• msg.Ack() 必须在处理完成后调用，如果处理失败可以调用 msg.Nak() 让消息重新消费。

7.3 消息确认模式的选型

JetStream 的 AckPolicy 有四种：AckNone、AckAll、AckExplicit、AckNotSet。最常用的是 AckExplicit。AckAll 适合批处理场景，它可以一次确认之前所有消息，降低 Ack 频次，但当处理中途失败时，之前已处理但未 Ack 的消息会被重新消费，可能有重复。AckNone 就是纯“阅后即焚”，尽量不要在生产环境用。

8. 消费者组：多个实例分摊消息

8.1 队列消费者

NATS 原生支持 Queue Subscribe，JetStream 也支持，但实现方式有差异。上面 Pull 模式创建的 Consumer 本身就可以被多个客户端订阅吗？实际上一个 Pull Consumer 在同一时刻可以被多个 PullSubscription 调用 Fetch，但服务端会为多个 Pull 请求分配不同批次，达到负载均衡效果。

不过更常见的做法是：为每个服务部署同一个 Consumer 名称，多实例共享消费进度和消息。比如创建 Durable 名为 “worker-group” 的 Consumer，三个服务实例都用它消费，即实现了“消费组”的效果，每条消息只被其中一个实例处理。

_, err := js.AddConsumer(streamName, &nats.ConsumerConfig{ Durable: "worker-group", AckPolicy: nats.AckExplicitPolicy, DeliverGroup: "my-queue-group", })

这里把 DeliverGroup 设为 “my-queue-group” 就能让多个订阅者形成队列组，实现负载均衡。

8.2 独立消费者

如果希望每个客户端都收到同一条消息，就不能用队列组，每个客户端用不同的 Consumer 名称订阅即可。JetStream 的每个 Consumer 都维护自己的消费进度，所以多个 Consumer 可以独立消费同一个 Stream 的同一批消息。

9. 消息持久化与重放：实战演示

9.1 如何验证消息确实持久化了

写完生产者和消费者后，可以做一个简单实验。先启动消费者，再启动生产者发 5 条消息，处理完后关闭消费者。然后重新启动消费者，你会发现之前已经被 Ack 的消息不会重复消费，但可以设计一个不 Ack 的场景，或者直接查看 Stream 里的消息数量来验证持久化。

用命令行查看：

nats stream list nats stream view order-stream

在 Go 里查看 Stream 信息：

info, err := js.StreamInfo("order-stream") if err != nil { log.Fatal(err) } log.Printf("Stream 消息数: %d, 字节数: %d", info.State.Messages, info.State.Bytes)

9.2 从指定序列号开始消费

JetStream 非常实用的能力是按需重放。假设消费者在处理到第 200 条消息时因为业务 Bug 挂掉了，但它已经把前 100 条消息 Ack 掉了，现在你想让它从第 150 条开始重新消费，就可以在 ConsumerConfig 里指定 OptStartSeq 为 150。若按时间重放，用 OptStartTime，比如从 15 分钟前开始。

config := &nats.ConsumerConfig{ Durable: "order-replay", AckPolicy: nats.AckExplicitPolicy, OptStartSeq: 150, }

这个能力在故障修复、数据补偿、测试回放中非常实用，Kafka 要实现类似效果通常需要重置 Consumer Group 的 offset，步骤繁琐且风险不小。

9.3 保留策略相互结合的实际效果

当 Stream 使用 LimitsPolicy 且设置了 MaxAge、MaxBytes、MaxMsgs 等限制时，消息会按照最先达到的条件被清理。这时候需要注意：即使是未被消费的消息，如果超过了保留时间/大小限制，一样会被删除。所以如果业务要求“必须保证至少消费成功一次”，要确保消费速度快于清理速度。

10. 常见问题与排查技巧实录

10.1 连接报错：nats: no responders available for request

这个错误多见于创建 Stream 或 Consumer 时。原因一般是 JetStream 没启用，或者客户端连接的不是 JetStream 的服务器。先确认服务器启动参数里有没有 -js，或者用 nats server report jetstream 查看状态。如果用了自定义端口，注意连接地址是否正确。

10.2 Fetch 一直返回超时，但消息明明已经发布

这种问题通常是因为 PullSubscribe 参数没有正确绑定。比如命令行创建了 Stream，Consumer 代码里 PullSubscribe 的 subject 写错了，就会导致没有消息。另一种常见情况是 ConsumerConfig 里 AckPolicy 设置了 AckNone，但你还是调用了 msg.Ack()，虽然不会报错，但状态管理会混乱。

检查时先看 nats stream view 里 Stream 有没有消息，再看 Consumer 的 Delivered 和 Acked 指标，如果 Delivered 在涨但 Acked 不动，说明消费端没有正确 Ack。

10.3 消费重复消息

JetStream 保证至少一次投递，所以业务处理要有幂等性。重复出现的原因包括：消费者 Ack 超时、客户端处理完但连接断掉导致 Ack 未达服务端、服务重启时未提交 Ack 的消息会被重新投递。解决方案是给消息加唯一 ID，落地数据库时做唯一约束，或者用 Redis SETNX 做去重。

10.4 磁盘占用快速上涨

Stream MaxBytes 和 MaxAge 是主要控制手段。如果磁盘还是持续上涨，别忘了 JetStream 在删除旧消息后，文件占用的空间不一定立刻释放（取决于内部压缩策略）。这种时候可以在低峰期重建 Stream，或者为 Stream 设置单独的存储目录，方便清理和维护。

10.5 消费端处理速度跟不上生产速度

优先检查消费端逻辑有没有串行阻塞。如果单条消息处理耗时很长，例如调用外部 HTTP 接口超时，就会阻塞 Fetch 循环。建议在 Fetch 内部使用 goroutine 并发处理，但要注意并发度，别一下子把消息全部拉进内存，否则可能导致 OOM。更好的方式是使用 JetStream 的分流，或者维护多个相同配置的 Consumer（用队列组分摊）。

10.6 首次 Fetch 时未消费到积压消息

新创建的 Consumer 默认从“当前最新位置”开始消费吗？其实不是，JetStream 新增 Consumer 时默认从该 Consumer 创建时刻的下一条消息开始。如果需要消费 Stream 中已有的全部历史消息，需要显式设置 DeliverPolicy。

config := &nats.ConsumerConfig{ DeliverPolicy: nats.DeliverAllPolicy, }

这个坑很容易踩，特别是你先启动了消费者，再发送积压消息时，如果消费者创建时间早于消息发布，默认从创建 Consumer 时刻开始消费，那么之前的消息就不会被送入消费者。好在 Durable Consumer 一旦创建并消费到新位置，后续重启会从最后一次 Ack 的位置继续，不会从头开始。

11. 经验总结与踩坑心得

11.1 使用 JetStream 时的核心习惯

我在实际项目里使用 JetStream 已经有一段时间，总结下来几个比较重要的习惯：

• 所有 Stream 和 Consumer 都通过代码初始化，并且做成幂等逻辑，这样不管是本地开发、测试环境还是生产环境，部署时不用去手动点击控制台，跑一遍初始化代码配置就齐全了。
• 所有重要业务消息都尽量带上唯一业务 ID，消费端做幂等，绝不寄希望于“JetStream 一定不会重复投递”。
• Push 和 Pull 的选择不要只看“方便”，如果客户端有可能是短生命周期，比如定时任务容器，尽量用 Pull 模式，每次启动后从某个时间点或序列号开始拉取，避免启动瞬间大量消息涌入。

11.2 一套可复用的项目目录建议

可以按这种方式组织 Go 项目：

jetstream-demo/ ├── go.mod ├── main.go ├── publisher/ │ └── publish.go ├── consumer/ │ └── subscribe.go └── common/ └── config.go

common 里放连接管理和配置读取，publisher 和 consumer 各自只做一件事：发送或消费。

11.3 再分享一个小技巧

调试 JetStream 配置时，可以用命令行工具的--force来重建 Stream，但注意这会清空旧消息。建议在测试环境测试保留策略，不要直接在业务 Stream 上操作。当时我为了省事，在生产环境直接改了 Stream 的 MaxAge，结果消息被大量删除，等到排查问题时才发现。生产环境任何配置变更，都要先评估对已有消费链路的影响。