深入解析 nats.go 遗留 JetStream API从消息发布、消费者订阅到流管理的完整 Go 实战指南【免费下载链接】inngestThe leading workflow orchestration platform. Run stateful step functions and AI workflows on serverless, servers, or the edge.项目地址: https://gitcode.com/GitHub_Trending/in/inngest在 NATS 生态中JetStream 为 NATS 引入了持久化消息流能力而 nats.go 客户端在nc.JetStream()上提供的这套 JetStream 上下文 API长期以来是 Go 开发者接入 JetStream 的最主要方式。本文以仓库内随附的 legacy_jetstream.md 文档为主体结合 js.go 与 jsm.go 中的实际源码实现系统讲解 JetStream 的发布、订阅与流管理三大能力并阐明遗留 API 与新版 jetstream 包的关系与取舍帮助读者正确选择 API 并写出可上线的生产级代码。一、背景什么是 JetStream什么是 nats.go 的遗留 APIJetStream 是构建在 NATS 之上的持久化消息流系统它允许消息被存储、重放和按消费者语义消费是事件驱动架构与工作流编排类应用例如本仓库 inngest 这类工作流编排平台常用的底层消息基础设施。nats.go 客户端库中nc.JetStream()返回的JetStreamContext接口封装了与 NATS Server 内部 JetStream API 的交互让开发者可以用普通 Go 函数调用的方式完成流的创建、消息的发布与订阅。需要特别说明的是从源码注释可以明确看到这套 API 被标记为遗留legacy接口。在 js.go 的JetStream()方法注释中写明JetStreamContext is part of legacy API. Users are encouraged to switch to the new JetStream API for enhanced capabilities and simplified API. Please refer to thejetstreampackage.即官方推荐新项目直接使用 jetstream 子包提供的新一代 API。但遗留 API 至今仍在大量存量代码中使用理解它的用法对维护既有系统、阅读开源项目源码依然至关重要。本文讲解的正是这套遗留 API 的完整用法同时会给出与新 API 的对比帮助读者做出选择。二、JetStream 基础用法连接、上下文与三种发布/订阅模式2.1 建立连接与创建 JetStream 上下文所有 JetStream 操作都从建立一个 NATS 连接开始然后通过nc.JetStream()创建 JetStream 上下文import github.com/nats-io/nats.go // Connect to NATS nc, _ : nats.Connect(nats.DefaultURL) // Create JetStream Context js, _ : nc.JetStream(nats.PublishAsyncMaxPending(256))nats.DefaultURL即nats://127.0.0.1:4222是本地默认连接地址。nc.JetStream()接收可变数量的JSOpt选项。从 js.go 的源码可以看到创建上下文时会设置一组默认值preAPI 请求前缀defaultAPIPrefix即$JS.APIwaitAPI 请求超时defaultRequestWaitmaxpa异步发布在途上限defaultAsyncPubAckInflight。随后逐个应用传入的JSOpt选项完成定制。这里用到的nats.PublishAsyncMaxPending(256)会将maxpa覆盖为 256。其实现位于 js.go当传入值小于 1 时会返回errors.New(nats: max ack pending should be 1)也就是说该选项必须大于等于 1。JetStreamContext本身并不直接代表连接或会话它是对底层*nats.Conn的轻量封装用于构造 JetStream 请求并接收服务端响应。2.2 同步发布最直接的写路径// Simple Stream Publisher js.Publish(ORDERS.scratch, []byte(hello))js.Publish(subj, data)是同步发布方法它把数据写入指定 subject并等待 JetStream 服务端返回PubAck确认Ack后才返回。在 js.go 中Publish最终委托给PublishMsg而PublishMsg的实现js.go展示了同步发布的关键细节若未显式设置超时ttl或上下文ctx默认使用js.opts.wait创建上下文时的请求等待时间同时设置ctx和ttl会返回ErrContextAndTimeout冲突错误发布重试默认值rwait为DefaultPubRetryWait250 毫秒rnum为DefaultPubRetryAttempts2 次定义于 js.go。同步发布语义简单、可靠适合对吞吐要求不高但需要逐个确认的场景。注意发布的目标 subject 必须被某个 Stream 的Subjects匹配覆盖否则 JetStream 会返回错误——这一点在本文第三部分创建 Stream 时会再次强调。2.3 异步发布高吞吐场景的正确姿势// Simple Async Stream Publisher for i : 0; i 500; i { js.PublishAsync(ORDERS.scratch, []byte(hello)) } select { case -js.PublishAsyncComplete(): case -time.After(5 * time.Second): fmt.Println(Did not resolve in time) }js.PublishAsync是异步发布方法它不等待服务端 Ack 就立即返回一个PubAckFuture适合批量发送高吞吐消息。随后通过js.PublishAsyncComplete()返回的 channel 等待所有在途异步消息完成确认。从 js.go 的PublishMsgAsync实现可以看出其内部机制每条异步消息会注册一个 pending Ack futureregisterPAF并统计在途数量numPending与上限maxPending流量控制Stall当在途消息数达到PublishAsyncMaxPending设定的上限时发布会进入阻塞等待默认 stall 等待时间为 200 毫秒defaultStallWait见 js.go。若超过上限仍无法发送会返回nats: stalled with too many outstanding async published messages错误。这就是为什么示例中将PublishAsyncMaxPending设为 256500 条消息中在途最多 256 条其余排队避免内存无限增长异步发布不接受超时或 context 参数设置它们会返回ErrContextAndTimeout。对于生产环境还应配合nats.PublishAsyncErrHandlerjs.go设置错误回调处理发送失败的消息MsgErrHandler会回传原始消息与错误可用于重发。注意PublishAsyncComplete只保证确认完成或超时真正的错误处理必须依赖PublishAsyncErrHandler。2.4 三种订阅模式异步、同步与拉取1异步推送订阅Push 模式 回调// Simple Async Ephemeral Consumer js.Subscribe(ORDERS.*, func(m *nats.Msg) { fmt.Printf(Received a JetStream message: %s\n, string(m.Data)) })js.Subscribe(subj, cb)注册一个回调处理器消息到达时由库内部派发。这种临时Ephemeral消费者没有指定Durable名称服务端会为其生成随机名称客户端退出后即被清理。2同步持久订阅Push 模式 手动拉取// Simple Sync Durable Consumer (optional SubOpts at the end) sub, err : js.SubscribeSync(ORDERS.*, nats.Durable(MONITOR), nats.MaxDeliver(3)) m, err : sub.NextMsg(timeout)js.SubscribeSync返回一个可同步调用的订阅sub.NextMsg(timeout)会阻塞至多timeout时间等待下一条消息。nats.Durable(MONITOR)指定持久消费者名称js.gonats.MaxDeliver(3)设置最大投递次数js.go超过该次数消息将进入死信处理流程。关于持久消费者的语义js.go 的Subscribe注释给出了权威说明如果指定了Durable()选项库会先尝试查找同名消费者找到则直接绑定不删除找不到则创建该持久消费者调用Unsubscribe()或Drain()后库会删除这个 JetStream 消费者如果使用Bind()选项则只查找并绑定查找失败会返回错误。3拉取订阅Pull 模式// Simple Pull Consumer sub, err : js.PullSubscribe(ORDERS.*, MONITOR) msgs, err : sub.Fetch(10)js.PullSubscribe(subj, durable)是拉取模式客户端按需调用sub.Fetch(10)主动批量拉取最多 10 条消息适合工作队列式的按需处理场景。从 js.go 的实现看当传入非空durable时它会自动追加Durable(durable)选项因此 Pull 订阅天然与持久消费者绑定。订阅模式兼容性校验processConsInfojs.go会校验订阅与消费者配置是否匹配——例如用 Pull 订阅绑定一个带DeliverSubject的 Push 消费者会返回ErrPullSubscribeToPushConsumer反之则返回ErrPullSubscribeRequired。2.5 退订与排空Unsubscribe vs Drain// Unsubscribe sub.Unsubscribe() // Drain sub.Drain()两者的区别定义于 nats.go 与 nats.goUnsubscribe()立即取消订阅后续消息不再投递可能丢失已发送但未处理的消息Drain()优雅排空——停止接收新消息但允许已在途/已排队消息处理完成后再关闭是生产环境推荐的收尾方式。三、JetStream 基础管理Stream 与 Consumer 的增删改查除消息收发外遗留 API 还提供了一套完整的流管理方法位于 jsm.go。下面按生命周期顺序展开。3.1 创建 Stream// Connect to NATS nc, _ : nats.Connect(nats.DefaultURL) // Create JetStream Context js, _ : nc.JetStream() // Create a Stream js.AddStream(nats.StreamConfig{ Name: ORDERS, Subjects: []string{ORDERS.*}, })js.AddStream(cfg)jsm.go向 JetStream 提交创建流的请求。StreamConfig中Name是流的唯一名称Subjects是该流捕获消息的主题模式。上面的示例意味着所有发往ORDERS.*如ORDERS.scratch、ORDERS.new的消息都会被ORDERS流持久化——这正是第二部分示例中发布到ORDERS.scratch能被存储的前提。StreamConfig还支持大量可选字段jsm.go常用的包括字段JSON 字段说明Retentionretention消息保留策略决定何时删除已消费消息MaxConsumersmax_consumers允许的最大消费者数MaxMsgsmax_msgs流最多存储的消息条数MaxBytesmax_bytes流最多存储的字节总量MaxAgemax_age消息最大保留时长MaxMsgsPerSubjectmax_msgs_per_subject每个 subject 最多保留的消息数Discarddiscard达到限制时的丢弃策略配合DiscardNewPerSubject使用Storagestorage存储后端类型内存/文件Replicasnum_replicas集群模式下流副本数NoAckno_ack是否禁用对接收消息的确认SubjectTransformsubject_transform消息存储时对 subject 做转换3.2 更新 Stream// Update a Stream js.UpdateStream(nats.StreamConfig{ Name: ORDERS, MaxBytes: 8, })js.UpdateStreamjsm.go用新的配置更新已有流典型用途是动态调整流限制如设置MaxBytes限制流的总存储字节数。注意更新时必须提供完整的期望配置至少包含Name并且流已存在若流不存在会返回错误。示例中MaxBytes: 8仅用于演示 API 形式真实场景应结合业务量设置合理值避免过小导致消息被快速丢弃。3.3 创建 Consumer// Create a Consumer js.AddConsumer(ORDERS, nats.ConsumerConfig{ Durable: MONITOR, })js.AddConsumer(stream, cfg)jsm.go在指定流上创建消费者。ConsumerConfig中Durable指定持久消费者名称。消费者是 JetStream 的消费游标它独立记录每个消费者已确认的消息位置因此多个消费者可以独立消费同一个流而互不干扰。补充说明第二部分示例中js.Subscribe、js.SubscribeSync、js.PullSubscribe传入的Durable选项本质上是订阅时按需创建持久消费者的便捷方式而AddConsumer则是显式、先行的管理操作两者创建出的消费者可以被对方绑定复用。3.4 删除 Consumer 与 Stream// Delete Consumer js.DeleteConsumer(ORDERS, MONITOR) // Delete Stream js.DeleteStream(ORDERS)js.DeleteConsumer(stream, consumer)jsm.go删除指定消费者js.DeleteStream(name)jsm.go删除整个流及其全部消息与消费者。这两个操作都是破坏性的生产环境中删除 Stream 前务必确认数据已无需保留。除增删改外jsm.go 还提供StreamInfojsm.go与ConsumerInfojsm.go用于查询流与消费者的当前状态是监控与调试的重要入口。四、遗留 API 与新 APIjetstream 包的关系与迁移仓库随附的 legacy_jetstream.md 开头明确指出当前 API 的 README 在 jetstream/README.md。也就是说nats.go 的 JetStream 能力分为两代维度遗留 API本文主题新 APIjetstream 包入口nc.JetStream()js.gojetstream.New(nc)代码位置js.go 与 jsm.gojetstream 子包接口风格大量*指针返回、opts函数式选项类型化、链式、错误处理更严格能力发布/订阅/流管理齐全在旧 API 基础上增强如 KV、对象存储、有序消费者、更细的错误类型官方态度标记为 legacy维护兼容推荐新项目使用从 js.go 的注释可以看出官方明确鼓励迁移而新 API 的选项命名也延续了旧 API 的语义例如WithPublishAsyncMaxPending对应旧 API 的PublishAsyncMaxPending参见 jetstream_options.goWithPublishAsyncErrHandler对应PublishAsyncErrHandlerjetstream_options.go迁移时概念可以平滑映射。五、可运行的完整示例与生产实践建议将本文全部要点汇总为一个可编译运行的完整示例假设本地已启动 NATS Server 并启用 JetStream默认nats://127.0.0.1:4222package main import ( fmt time github.com/nats-io/nats.go ) func main() { // 1. 连接与上下文 nc, err : nats.Connect(nats.DefaultURL) if err ! nil { panic(err) } defer nc.Close() js, err : nc.JetStream(nats.PublishAsyncMaxPending(256)) if err ! nil { panic(err) } // 2. 管理创建流 if _, err : js.AddStream(nats.StreamConfig{ Name: ORDERS, Subjects: []string{ORDERS.*}, }); err ! nil { panic(err) } // 3. 发布同步 异步 if _, err : js.Publish(ORDERS.scratch, []byte(hello)); err ! nil { panic(err) } for i : 0; i 500; i { js.PublishAsync(ORDERS.scratch, []byte(hello)) } select { case -js.PublishAsyncComplete(): case -time.After(5 * time.Second): fmt.Println(Did not resolve in time) } // 4. 订阅异步临时消费者 if _, err : js.Subscribe(ORDERS.*, func(m *nats.Msg) { fmt.Printf(Received a JetStream message: %s\n, string(m.Data)) m.Ack() }); err ! nil { panic(err) } // 5. 订阅同步持久消费者 sub, err : js.SubscribeSync(ORDERS.*, nats.Durable(MONITOR), nats.MaxDeliver(3)) if err ! nil { panic(err) } m, err : sub.NextMsg(2 * time.Second) if err nil { m.Ack() } // 6. 订阅拉取消费者 pullSub, err : js.PullSubscribe(ORDERS.*, MONITOR) if err ! nil { panic(err) } msgs, err : pullSub.Fetch(10) if err nil { for _, msg : range msgs { msg.Ack() } } // 7. 收尾 pullSub.Drain() sub.Drain() js.DeleteConsumer(ORDERS, MONITOR) js.DeleteStream(ORDERS) }生产环境落地时以下几点值得特别注意务必处理错误本文示例为了简洁多省略了错误判断但nc.JetStream()返回的错误表示选项配置不一致如PublishAsyncMaxPending小于 1Publish、AddStream等调用的错误直接反映服务端拒绝原因都应显式处理异步发布必须配置错误处理器PublishAsyncErrHandler是异步发布可靠性的最后防线在途消息失败只能通过它捕获并决定是否重发按需选择消费模式任务分发、按需批量处理用 PullFetch低延迟实时流式处理用 PushSubscribe/SubscribeSync需要游标独立、断点续传用Durable持久消费者收尾使用Drain()而非直接Unsubscribe()避免丢弃在途消息新项目优先评估 jetstream 新 API存量项目可基于本文的语义映射逐步迁移。六、总结本文完整梳理了 nats.go 遗留 JetStream API 的全部核心能力通过nc.JetStream()建立上下文用Publish/PublishAsync完成同步与异步发布含在途上限与 stall 流量控制机制用Subscribe/SubscribeSync/PullSubscribe覆盖推送、同步、拉取三种消费模式用AddStream/UpdateStream/AddConsumer/DeleteConsumer/DeleteStream完成流与消费者的全生命周期管理并厘清了它与新 jetstream 包的演进关系。无论你是维护既有 NATS 代码还是评估 JetStream 客户端 API 的选型这套遗留 API 的知识都是理解 nats.go 生态不可或缺的一环。【免费下载链接】inngestThe leading workflow orchestration platform. Run stateful step functions and AI workflows on serverless, servers, or the edge.项目地址: https://gitcode.com/GitHub_Trending/in/inngest创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考