natsx

go-toolbox/pkg/natsx

一个功能丰富、高性能的 Go 语言 NATS 客户端易用性封装库，提供泛型事件订阅、广播、批量流式消费和消费者池管理

📖 特性 & 文档导航

📖 完整文档：查看 文档中心 了解所有功能和学习路径

📦 安装

go get github.com/HitGameAI/apex-toolbox/pkg/natsx

🚀 快速开始

import (
    "github.com/nats-io/nats.go"
    natsx "github.com/HitGameAI/apex-toolbox/pkg/natsx"
)

// 1. 创建 NATS 连接（由调用方管理连接生命周期）
conn, _ := nats.Connect("nats://127.0.0.1:4222")

// 2. 创建客户端
client, _ := natsx.NewClient(conn)
defer client.Close()

// 3. 发布事件
natsx.PublishEvent(client, "user.created", &UserCreated{Name: "张三"})

// 4. 订阅事件（泛型自动反序列化）
natsx.Subscribe[UserCreated](client, "user.created", "order-service", func(evt *UserCreated) error {
    fmt.Println("User created:", evt.Name)
    return nil
})

💡 详细教程：查看 📘 快速入门文档 了解完整的安装和使用步骤

🏗️ 核心特性

泛型事件订阅

// 普通订阅 - QueueSubscribe 负载均衡
natsx.Subscribe[OrderEvent](client, "order.created", "payment-service", func(evt *OrderEvent) error {
    return processPayment(evt)
})

// 广播订阅 - 所有订阅者都收到消息
natsx.SubscribeBroadcast[UserEvent](client, "user.updated", func(evt *UserEvent) error {
    return refreshCache(evt)
})

// 批量流式消费 - JetStream Pull 模式
natsx.SubscribeStreamBatch[LogEvent](client, "logs.batch", "analytics", func(evts []*LogEvent) error {
    return batchInsert(evts)
}, natsx.WithBatchSize(100), natsx.WithMaxWait(5*time.Second))

消费者池

// 全局消费者池 - 所有订阅共享
client.InitWorkerPool(10, 1000)
natsx.Subscribe[Event](client, "topic", "svc", handler, natsx.WithIntoGlobalPool())

// 局部消费者池 - 每个订阅独立
natsx.Subscribe[Event](client, "topic", "svc", handler, natsx.WithLocalPoolSize(5, 200))

JetStream 支持

client.EnableJetStream()

// 发布持久化消息
ack, _ := client.PublishJetStream(ctx, "order.created", data)

// 订阅 JetStream 消息（自动 ACK）
natsx.Subscribe[OrderEvent](client, "order.created", "svc", handler,
    natsx.WithMaxAckWait(30*time.Second),
    natsx.WithMsgMaxRetry(5),
)

重试与应答决策

handler 返回错误时，库按应答决策表自动选择 Term / NakWithDelay / Nak：

// 永久性失败：handler 包装 ErrPermanent 哨兵，库直接 Term 终止消息
natsx.Subscribe[OrderEvent](client, "order.paid", "svc", func(ctx context.Context, evt *OrderEvent) error {
    if evt.OrderNo == "" {
        return fmt.Errorf("%w: order not found", natsx.ErrPermanent) // 重试不可修复
    }
    return processPayment(ctx, evt) // 临时错误：自动退避重投
},
    // 指数退避：1s 起步，逐次 ×2，封顶 30s，叠加随机抖动避免重投风暴同步
    natsx.WithRetryBackoff(natsx.Backoff{Base: time.Second, Max: 30 * time.Second, Factor: 2, Jitter: true}),
    natsx.WithMsgMaxRetry(5), // 最多投递 5 次后 Term
    // natsx.WithUnlimitedDelivery(), // 或：无限重投（除 ErrPermanent 外永不放弃）
)

上下文传播（ContextInjector）

每条消息处理前从消息 Header 继承跨服务上下文（如 trace_id），依赖倒置设计，库自身不感知具体实现：

// 注入器：从消息 Header 恢复 trace 上下文（桥接网关中间件的 trace 逻辑）
injector := func(ctx context.Context, msg *nats.Msg) context.Context {
    if traceID := msg.Header.Get("X-Trace-Id"); traceID != "" {
        ctx = gwMiddleware.WithTraceID(ctx, traceID)
    }
    return ctx
}

natsx.Subscribe[OrderEvent](ctx, client, "order.paid", "svc", handler,
    natsx.WithContextInjector(injector),
    natsx.WithMaxAckWait(30*time.Second), // 消息级 ctx deadline 与 AckWait 对齐，
                                           // 消除「handler 仍在跑但 JetStream 已重投」的双活窗口
)

📖 详细说明：查看 🏊 消费者池 和 🔗 JetStream 了解更多配置

🧪 测试

测试内嵌进程内 NATS 服务器（启用 JetStream），不依赖任何外部环境，直接运行：

# 运行全部测试（覆盖率 100%）
go test ./... -cover

# 连接外部服务器（可选，CI 复用常驻实例时设置）
NATS_TEST_URL=nats://127.0.0.1:4222 go test ./... -v

# 运行指定测试
go test -v -run TestSubscribe

📚 相关资源

• 📖 完整文档中心 - 所有功能文档和学习路径
• 🐛 问题反馈 - 报告 bug 或提出建议
• 💬 讨论区 - 技术交流

📦 依赖

• nats.go - NATS 官方 Go 客户端
• go-toolbox - WorkerPool 和重试工具
• go-toolbox/pkg/logger - 结构化日志
• json-iterator - 高性能 JSON 序列化

🤝 贡献

1. Fork 本仓库
2. 创建特性分支 (git checkout -b feature/amazing-feature)
3. 提交更改 (git commit -m '✨ feat: Add amazing feature')
4. 推送到分支 (git push origin feature/amazing-feature)
5. 开启 Pull Request

📋 Git Commit Emoji 规范

📄 许可证

MIT License - 详见 LICENSE

👨‍💻 作者

Kamal Yang (@kamalyes)