Go 客户端实战指南

开发环境连接信息

NATS Server: nats://localhost:4222

1. 依赖安装

bash

# 安装 nats.go 客户端库（最新稳定版）
go get github.com/nats-io/nats.go@latest

# 查看当前版本
go list -m github.com/nats-io/nats.go

版本说明：

版本范围	JetStream API	说明
< v1.20	旧版（nats.JetStream()）	不推荐，功能受限
v1.20 ~ v1.35	新版（jetstream.New()）Beta	逐步稳定
>= v1.36	新版（jetstream.New()）推荐	功能完整，API 稳定

本文所有示例基于 nats.go v1.49.0 和 新版 JetStream API（jetstream 包）。

// go.mod 示例
module myapp

go 1.21

require (
    github.com/nats-io/nats.go v1.49.0
)

2. 连接管理

2.1 基础连接

package main

import (
    "log"

    "github.com/nats-io/nats.go"
)

func basicConnect() {
    // 最简单的连接方式
    // 开发环境地址：nats://localhost:4222
    nc, err := nats.Connect("nats://localhost:4222")
    if err != nil {
        log.Fatal("连接失败:", err)
    }
    defer nc.Close()

    log.Printf("已连接到 NATS，服务器: %s", nc.ConnectedUrl())
}

2.2 连接选项

func connectWithOptions() {
    nc, err := nats.Connect(
        "nats://localhost:4222",

        // 连接名称（方便在监控中识别）
        nats.Name("my-service-prod-1"),

        // 认证（如果服务器要求）
        nats.UserInfo("username", "password"),
        // 或使用 Token 认证
        nats.Token("my-secret-token"),
        // 或使用 NKey
        // nats.NkeyOptionFromSeed("SUAM..."),

        // 重连配置
        nats.MaxReconnects(-1),           // -1 表示无限重连（推荐生产环境）
        nats.ReconnectWait(2*time.Second), // 重连间隔（默认 2s）
        nats.ReconnectJitter(
            500*time.Millisecond,  // 非 TLS 连接的随机抖动范围
            2*time.Second,         // TLS 连接的随机抖动范围
        ),

        // Ping 配置（检测连接存活）
        nats.PingInterval(30*time.Second), // Ping 间隔（默认 2min）
        nats.MaxPingsOutstanding(3),        // 最多允许多少个未响应的 Ping

        // 缓冲区配置
        nats.ReconnectBufSize(8*1024*1024), // 重连期间的消息缓冲区，默认 8MB

        // 连接超时
        nats.Timeout(5*time.Second),
    )
    if err != nil {
        log.Fatal("连接失败:", err)
    }
    defer nc.Close()
}

2.3 连接事件回调

func connectWithCallbacks() {
    nc, err := nats.Connect(
        "nats://localhost:4222",
        nats.Name("event-service"),
        nats.MaxReconnects(-1),

        // 连接断开时触发（还在尝试重连）
        nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
            if err != nil {
                log.Printf("[DISCONNECT] 与 NATS 连接断开: %v", err)
            } else {
                log.Println("[DISCONNECT] 与 NATS 连接正常断开")
            }
            // 在此期间，发布的消息会缓冲在 ReconnectBufSize 指定的缓冲区中
            // 重连成功后会自动重发（Core NATS 无持久化，JetStream 消息有 PubAck 确认）
        }),

        // 重连成功时触发
        nats.ReconnectHandler(func(nc *nats.Conn) {
            log.Printf("[RECONNECT] 已重新连接到: %s", nc.ConnectedUrl())
            // 可以在这里重新订阅或执行恢复逻辑
        }),

        // 连接最终关闭（超过最大重试次数，或显式调用 Close()/Drain()）
        nats.ClosedHandler(func(nc *nats.Conn) {
            log.Println("[CLOSED] NATS 连接已永久关闭")
            // 可以在这里发送告警或触发优雅退出
        }),

        // 异步错误（例如慢消费者被服务器强制关闭）
        nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) {
            if err == nats.ErrSlowConsumer {
                pending, _, _ := sub.Pending()
                log.Printf("[ERROR] 慢消费者告警: Subject=%s, 积压消息数=%d",
                    sub.Subject, pending)
            } else {
                log.Printf("[ERROR] NATS 异步错误: %v", err)
            }
        }),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Drain()
}

2.4 多节点连接（集群高可用）

func connectToCluster() {
    // 方式 1：逗号分隔的 URL 列表（客户端随机选一个连接，失败后尝试下一个）
    nc, err := nats.Connect(
        "nats://nats-1:4222,nats://nats-2:4222,nats://nats-3:4222",
        nats.MaxReconnects(-1),
        nats.ReconnectWait(1*time.Second),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Drain()

    log.Printf("已连接到集群节点: %s", nc.ConnectedUrl())
    // 注意：连接到任意一个节点后，NATS 会自动推送集群路由信息
    // 断开后会自动尝试连接其他已知节点
}

2.5 TLS 连接

func connectWithTLS() {
    // 方式 1：使用 TLS 配置结构
    tlsConfig := &tls.Config{
        // 生产环境：加载 CA 证书验证服务端
        RootCAs: loadCACertPool("/path/to/ca.pem"),
        // 双向 TLS（mTLS）：客户端证书
        Certificates: []tls.Certificate{loadClientCert(
            "/path/to/client.pem",
            "/path/to/client-key.pem",
        )},
        MinVersion: tls.VersionTLS12,
    }

    nc, err := nats.Connect(
        "tls://nats:4222",
        nats.Secure(tlsConfig),
        nats.Name("secure-client"),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Drain()

    // 方式 2：证书文件路径（便捷方式）
    nc2, err := nats.Connect(
        "tls://nats:4222",
        nats.ClientCert("/path/to/client.pem", "/path/to/client-key.pem"),
        nats.RootCAs("/path/to/ca.pem"),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer nc2.Drain()
}

func loadCACertPool(caFile string) *x509.CertPool {
    pool := x509.NewCertPool()
    data, _ := os.ReadFile(caFile)
    pool.AppendCertsFromPEM(data)
    return pool
}

func loadClientCert(certFile, keyFile string) tls.Certificate {
    cert, _ := tls.LoadX509KeyPair(certFile, keyFile)
    return cert
}

2.6 drain vs close 的区别

func drainVsClose(nc *nats.Conn) {
    // nc.Close()：立即断开连接
    // - 优点：立即完成
    // - 缺点：未处理的消息可能丢失（订阅缓冲区中的消息被丢弃）
    // - 适用：进程崩溃前的紧急关闭，或不关心消息丢失的场景
    nc.Close()

    // nc.Drain()：优雅排水关闭（推荐用于正常退出）
    // - 停止接受新消息（Unsubscribe 所有订阅）
    // - 等待已缓冲的消息处理完成
    // - 刷新发布缓冲区（确保所有 Publish 已发送给服务端）
    // - 最后关闭连接
    // - 适用：服务优雅停机、滚动升级
    err := nc.Drain()
    if err != nil {
        log.Printf("Drain 错误: %v", err)
    }

    // 使用信号处理实现优雅关闭示例
}

func gracefulShutdown(nc *nats.Conn) {
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, os.Interrupt, syscall.SIGTERM)

    <-quit
    log.Println("收到退出信号，开始优雅关闭...")

    // 设置 Drain 超时
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    doneCh := make(chan struct{})
    go func() {
        nc.Drain()
        close(doneCh)
    }()

    select {
    case <-doneCh:
        log.Println("优雅关闭完成")
    case <-ctx.Done():
        log.Println("优雅关闭超时，强制关闭")
        nc.Close()
    }
}

3. Core NATS 使用

Core NATS 提供最基础的发布/订阅功能，消息不持久化，适合实时事件、Command 分发、健康检查等场景。

3.1 Publish — 发布

func publishExamples(nc *nats.Conn) {
    // 基础发布（即发即忘，不等待确认）
    err := nc.Publish("events.user.login", []byte(`{"user_id": "123"}`))
    if err != nil {
        log.Printf("Publish 失败: %v", err)
        return
    }

    // 带消息头的发布（NATS 2.2+ 支持 Headers）
    msg := &nats.Msg{
        Subject: "events.order.created",
        Data:    []byte(`{"order_id": "ORD-001"}`),
        Header: nats.Header{
            "Content-Type":    []string{"application/json"},
            "X-Trace-ID":      []string{"trace-abc123"},
            "X-Source-Service": []string{"order-service"},
        },
    }
    if err := nc.PublishMsg(msg); err != nil {
        log.Printf("PublishMsg 失败: %v", err)
        return
    }

    // Flush 确保消息已发送到服务端（用于测试或需要确认时）
    if err := nc.Flush(); err != nil {
        log.Printf("Flush 失败: %v", err)
    }

    // FlushTimeout 带超时的 Flush
    if err := nc.FlushTimeout(3 * time.Second); err != nil {
        log.Printf("FlushTimeout 失败: %v", err)
    }
}

func subscribeExamples(nc *nats.Conn) {
    // 异步订阅（推荐，不阻塞）
    sub, err := nc.Subscribe("events.user.>", func(msg *nats.Msg) {
        // 此回调在独立 goroutine 中执行
        fmt.Printf("收到消息: subject=%s data=%s\n",
            msg.Subject, string(msg.Data))

        // 读取消息头
        if traceID := msg.Header.Get("X-Trace-ID"); traceID != "" {
            fmt.Printf("Trace-ID: %s\n", traceID)
        }
    })
    if err != nil {
        log.Printf("Subscribe 失败: %v", err)
        return
    }
    defer sub.Unsubscribe()

    // 订阅带通配符
    // * 匹配单个 token
    subStar, _ := nc.Subscribe("events.*.created", func(msg *nats.Msg) {
        fmt.Printf("新建事件: %s\n", msg.Subject)
    })
    defer subStar.Unsubscribe()

    // > 匹配多层
    subAll, _ := nc.Subscribe("events.>", func(msg *nats.Msg) {
        fmt.Printf("所有事件: %s\n", msg.Subject)
    })
    defer subAll.Unsubscribe()

    // 同步订阅（适合测试或需要精确控制流的场景）
    subSync, err := nc.SubscribeSync("events.health")
    if err != nil {
        log.Printf("SubscribeSync 失败: %v", err)
        return
    }
    defer subSync.Unsubscribe()

    // 同步接收（阻塞等待消息，带超时）
    msg, err := subSync.NextMsg(5 * time.Second)
    if err != nil {
        if err == nats.ErrTimeout {
            fmt.Println("5秒内没有收到消息")
        } else {
            log.Printf("NextMsg 失败: %v", err)
        }
        return
    }
    fmt.Printf("同步收到: %s\n", string(msg.Data))

    // 限制接收条数后自动 Unsubscribe
    autoUnsub, _ := nc.Subscribe("events.once", func(msg *nats.Msg) {
        fmt.Println("只处理一条消息")
    })
    autoUnsub.AutoUnsubscribe(1) // 接收 1 条后自动取消订阅

    // 设置订阅的挂起消息上限（防止慢消费者内存无限增长）
    sub.SetPendingLimits(10000, 10*1024*1024) // 10000条 或 10MB
}

3.3 QueueSubscribe — 负载均衡消费

func queueSubscribeExample(nc *nats.Conn) {
    // QueueSubscribe：多个消费者共享同一个队列，每条消息只投递给一个消费者
    // 实现 Core NATS 层面的负载均衡

    handler := func(msg *nats.Msg) {
        fmt.Printf("[worker] 处理消息: %s\n", string(msg.Data))
        // 模拟处理耗时
        time.Sleep(100 * time.Millisecond)
    }

    // 启动 3 个 worker，共享同一个队列 "payment-workers"
    for i := 1; i <= 3; i++ {
        workerID := i
        sub, err := nc.QueueSubscribe("payments.process", "payment-workers", func(msg *nats.Msg) {
            fmt.Printf("[worker-%d] 处理: %s\n", workerID, string(msg.Data))
        })
        if err != nil {
            log.Printf("QueueSubscribe 失败: %v", err)
            return
        }
        defer sub.Unsubscribe()
        _ = handler
    }

    // 发布 10 条消息，会被 3 个 worker 均匀消费（负载均衡）
    for i := 0; i < 10; i++ {
        nc.Publish("payments.process", []byte(fmt.Sprintf(`{"payment_id": %d}`, i)))
    }
    nc.Flush()
    time.Sleep(2 * time.Second)
}

3.4 Request/Reply — 请求响应

func requestReplyExample(nc *nats.Conn) {
    // --- 服务端（处理请求）---
    // Subscribe 接收请求，使用 msg.Respond 回复
    responder, _ := nc.Subscribe("rpc.user.get", func(msg *nats.Msg) {
        // 解析请求
        userID := string(msg.Data)
        fmt.Printf("查询用户: %s\n", userID)

        // 构造响应
        response := fmt.Sprintf(`{"id": "%s", "name": "张三", "email": "zhangsan@example.com"}`, userID)

        // 回复到 msg.Reply（由客户端自动生成的一次性 Subject）
        if err := msg.Respond([]byte(response)); err != nil {
            log.Printf("Respond 失败: %v", err)
        }
    })
    defer responder.Unsubscribe()

    // --- 客户端（发起请求）---
    // Request 会自动：
    // 1. 生成一个唯一的 ReplyTo Subject（类似 "_INBOX.xxx"）
    // 2. 订阅该 ReplyTo Subject
    // 3. 发布请求
    // 4. 等待响应（阻塞，带超时）
    // 5. 取消订阅并返回响应
    response, err := nc.Request("rpc.user.get", []byte("user-123"), 5*time.Second)
    if err != nil {
        if err == nats.ErrNoResponders {
            fmt.Println("没有服务处理此请求（服务未启动？）")
        } else if err == nats.ErrTimeout {
            fmt.Println("请求超时")
        } else {
            log.Printf("Request 失败: %v", err)
        }
        return
    }
    fmt.Printf("用户信息: %s\n", string(response.Data))

    // RequestMsg：可以携带 Headers 的请求
    req := &nats.Msg{
        Subject: "rpc.user.get",
        Data:    []byte("user-456"),
        Header: nats.Header{
            "X-Timeout-Ms": []string{"3000"},
            "X-Caller":     []string{"api-gateway"},
        },
    }
    resp, err := nc.RequestMsg(req, 5*time.Second)
    if err != nil {
        log.Printf("RequestMsg 失败: %v", err)
        return
    }
    fmt.Printf("响应: %s\n", string(resp.Data))
}

4. JetStream v2 API

从 nats.go v1.20 开始，推荐使用新版 jetstream 包（github.com/nats-io/nats.go/jetstream）。新版 API 更清晰、错误处理更完善、支持 Context。

4.1 获取 JetStream Context

import (
    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

func getJetStream(nc *nats.Conn) jetstream.JetStream {
    js, err := jetstream.New(nc)
    if err != nil {
        log.Fatal("创建 JetStream context 失败:", err)
    }
    return js
}

4.2 Stream 操作

func streamOperations(js jetstream.JetStream) {
    ctx := context.Background()

    // CreateOrUpdateStream：幂等创建（已存在则更新配置）
    stream, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
        Name:     "ORDERS",
        Subjects: []string{"orders.>"},
        Replicas: 3,
        Storage:  jetstream.FileStorage,
        MaxAge:   7 * 24 * time.Hour,
        MaxBytes: 10 << 30,
    })
    if err != nil {
        log.Fatal("创建 Stream 失败:", err)
    }
    fmt.Printf("Stream 已就绪: %s\n", stream.CachedInfo().Config.Name)

    // CreateStream：Stream 已存在时返回 ErrStreamNameAlreadyInUse
    _, err = js.CreateStream(ctx, jetstream.StreamConfig{Name: "NEW_STREAM"})
    if errors.Is(err, jetstream.ErrStreamNameAlreadyInUse) {
        fmt.Println("Stream 已存在")
    }

    // UpdateStream：更新已有 Stream 的配置（不能修改 Name、Storage 等不可变字段）
    _, err = js.UpdateStream(ctx, jetstream.StreamConfig{
        Name:    "ORDERS",
        MaxAge:  30 * 24 * time.Hour, // 延长保留时间
        MaxBytes: 50 << 30,
    })

    // Stream：获取 Stream handle（用于后续操作）
    s, err := js.Stream(ctx, "ORDERS")
    if err != nil {
        log.Printf("Stream 不存在: %v", err)
        return
    }

    // 获取 Stream 状态
    info, _ := s.Info(ctx) // 直接请求服务端，获取最新状态
    fmt.Printf("消息数: %d, 存储: %d MB\n",
        info.State.Msgs, info.State.Bytes/1024/1024)

    // 获取缓存状态（不发网络请求，可能略过期）
    cachedInfo := s.CachedInfo()
    fmt.Printf("Stream Leader: %s\n", cachedInfo.Cluster.Leader)

    // DeleteStream：删除 Stream（连同所有数据）
    // js.DeleteStream(ctx, "ORDERS") // 危险操作，谨慎使用

    // ListStreams：遍历所有 Stream
    iter := js.ListStreams(ctx)
    for s := range iter.Status() {
        fmt.Printf("Stream: %s 消息数: %d\n", s.Config.Name, s.State.Msgs)
    }
    if err := iter.Err(); err != nil {
        log.Printf("ListStreams 错误: %v", err)
    }
}

4.3 Publish — 发布消息到 Stream

func publishToStream(js jetstream.JetStream) {
    ctx := context.Background()

    // 普通同步发布（等待 PubAck，确认消息已持久化）
    ack, err := js.Publish(ctx, "orders.placed", []byte(`{"order_id":"ORD-001"}`))
    if err != nil {
        log.Printf("发布失败: %v", err)
        return
    }
    fmt.Printf("已持久化: stream=%s seq=%d duplicate=%v\n",
        ack.Stream, ack.Sequence, ack.Duplicate)

    // 带消息头发布
    msg := &nats.Msg{
        Subject: "orders.placed",
        Data:    []byte(`{"order_id":"ORD-002"}`),
        Header: nats.Header{
            "Nats-Msg-Id":     []string{"unique-id-002"}, // 去重 ID（配合 Duplicates 窗口）
            "Content-Type":   []string{"application/json"},
        },
    }
    ack, err = js.PublishMsg(ctx, msg)
    if err != nil {
        log.Printf("PublishMsg 失败: %v", err)
        return
    }
    if ack.Duplicate {
        fmt.Println("消息已存在（去重），不重复处理")
    }

    // 异步发布（高吞吐场景，不阻塞等待每条消息的 ACK）
    // 推荐方式：使用 PublishAsync + AcksFuture
    paf, err := js.PublishAsync("orders.placed", []byte(`{"order_id":"ORD-003"}`))
    if err != nil {
        log.Printf("PublishAsync 失败（本地错误）: %v", err)
        return
    }

    // 之后某个时间检查 ACK
    select {
    case ack := <-paf.Ok():
        fmt.Printf("异步 ACK 收到: seq=%d\n", ack.Sequence)
    case err := <-paf.Err():
        fmt.Printf("异步发布失败: %v\n", err)
    case <-time.After(5 * time.Second):
        fmt.Println("等待 ACK 超时")
    }
}

4.4 Consumer 操作与 Pull Consumer 消费

func pullConsumerExample(js jetstream.JetStream) {
    ctx := context.Background()

    // 创建或更新 Pull Consumer
    cons, err := js.CreateOrUpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
        Name:          "order-processor",
        Durable:       "order-processor", // Durable Name 让 Consumer 状态持久化
        Description:   "订单处理消费者",
        FilterSubject: "orders.placed",

        // 消费起点
        DeliverPolicy: jetstream.DeliverAllPolicy, // 从第一条消息开始

        // ACK 策略
        AckPolicy: jetstream.AckExplicitPolicy, // 需要显式 ACK

        // ACK 超时：消息投递后多久没有 ACK 就重投
        AckWait: 30 * time.Second,

        // 最大重投次数（超过后消息进入 Dead Letter 或被丢弃）
        MaxDeliver: 3,

        // 最大 ACK 挂起数（未 ACK 的消息上限，相当于并发控制）
        MaxAckPending: 1000,

        // 重投退避（exponential backoff）
        BackOff: []time.Duration{
            5 * time.Second,
            30 * time.Second,
            2 * time.Minute,
        },
    })
    if err != nil {
        log.Fatal("创建 Consumer 失败:", err)
    }

    // 方式 1：Fetch — 一次性拉取指定数量的消息（适合批处理）
    msgs, err := cons.Fetch(10, jetstream.FetchMaxWait(5*time.Second))
    if err != nil {
        log.Printf("Fetch 失败: %v", err)
        return
    }
    for msg := range msgs.Messages() {
        fmt.Printf("处理: %s\n", string(msg.Data()))
        // 处理成功后 ACK
        msg.Ack()
        // 或：处理失败后 NAK（立即重投）
        // msg.Nak()
        // 或：处理失败后延迟重投
        // msg.NakWithDelay(10 * time.Second)
        // 或：需要更多时间处理（延长 AckWait）
        // msg.InProgress()
        // 或：永久丢弃（不再重投）
        // msg.Term()
    }
    if err := msgs.Error(); err != nil {
        log.Printf("Fetch 遍历错误: %v", err)
    }

    // 方式 2：FetchNoWait — 只拉取当前可用消息，立即返回（不等待新消息）
    msgs2, err := cons.FetchNoWait(100)
    if err != nil {
        log.Printf("FetchNoWait 失败: %v", err)
        return
    }
    for msg := range msgs2.Messages() {
        msg.Ack()
    }

    // 方式 3：Messages() — 持续消费（长轮询，推荐用于常驻 Worker）
    iter, err := cons.Messages(
        jetstream.PullMaxMessages(50), // 每次最多预取 50 条
        jetstream.PullHeartbeat(5*time.Second), // 空闲时发送心跳保持连接
    )
    if err != nil {
        log.Printf("Messages 失败: %v", err)
        return
    }
    defer iter.Stop()

    for {
        msg, err := iter.Next()
        if err != nil {
            if errors.Is(err, jetstream.ErrMsgIteratorClosed) {
                fmt.Println("消息迭代器已关闭")
                return
            }
            log.Printf("Next 错误: %v", err)
            return
        }

        // 处理消息
        fmt.Printf("收到消息: seq=%d data=%s\n",
            msg.Metadata().Sequence.Stream,
            string(msg.Data()))

        // 处理成功
        msg.Ack()
    }
}

4.5 Push Consumer 消费

func pushConsumerExample(js jetstream.JetStream) {
    ctx := context.Background()

    // Push Consumer：服务端主动推送消息给客户端
    // 适合需要低延迟的场景（不需要主动 Fetch）
    cons, err := js.CreateOrUpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
        Name:           "order-notifier",
        Durable:        "order-notifier",
        FilterSubject:  "orders.>",
        DeliverPolicy:  jetstream.DeliverNewPolicy, // 只消费新消息
        AckPolicy:      jetstream.AckExplicitPolicy,
        AckWait:        10 * time.Second,
        MaxDeliver:     5,
    })
    if err != nil {
        log.Fatal("创建 Consumer 失败:", err)
    }

    // Consume：注册消息处理函数（异步回调）
    consCtx, err := cons.Consume(func(msg jetstream.Msg) {
        fmt.Printf("[push] 收到: %s\n", string(msg.Data()))

        meta, _ := msg.Metadata()
        fmt.Printf("  Stream Seq: %d, Consumer Seq: %d, Redelivered: %d\n",
            meta.Sequence.Stream, meta.Sequence.Consumer, meta.NumDelivered)

        msg.Ack()
    },
        jetstream.ConsumeErrHandler(func(consumeCtx jetstream.ConsumeContext, err error) {
            log.Printf("Consume 错误: %v", err)
        }),
    )
    if err != nil {
        log.Fatal("Consume 失败:", err)
    }
    defer consCtx.Stop()

    // 保持运行
    time.Sleep(30 * time.Second)
}

4.6 KV Store 操作（见第 07 篇完整示例）

func kvStoreExample(js jetstream.JetStream) {
    ctx := context.Background()

    // 创建 KV Bucket
    kv, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
        Bucket:   "APP_CONFIG",
        History:  5,
        Replicas: 3,
    })
    if err != nil {
        log.Fatal(err)
    }

    // 基本操作
    kv.Put(ctx, "db.host", []byte("localhost"))
    entry, _ := kv.Get(ctx, "db.host")
    fmt.Printf("db.host = %s (rev=%d)\n", string(entry.Value()), entry.Revision())

    // Watch
    watcher, _ := kv.WatchAll(ctx)
    defer watcher.Stop()
    // （详见第 07 篇）
}

5. 错误处理模式

5.1 常见错误类型

package errors_demo

import (
    "errors"
    "log"

    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

func handleCommonErrors(nc *nats.Conn, js jetstream.JetStream) {
    ctx := context.Background()

    // --- Core NATS 错误 ---

    // ErrNoResponders：发起 Request 时没有任何订阅者响应
    _, err := nc.Request("nonexistent.service", []byte("ping"), 3*time.Second)
    if errors.Is(err, nats.ErrNoResponders) {
        log.Println("服务不可用，没有处理者注册")
        // 处理：服务降级、返回错误给调用方、告警
        return
    }

    // ErrTimeout：等待超时
    _, err = nc.Request("slow.service", []byte("ping"), 1*time.Second)
    if errors.Is(err, nats.ErrTimeout) {
        log.Println("请求超时（1秒内无响应）")
        // 处理：重试（带退避），或返回超时错误
        return
    }

    // ErrConnectionClosed：连接已关闭
    nc.Close()
    err = nc.Publish("test", []byte("data"))
    if errors.Is(err, nats.ErrConnectionClosed) {
        log.Println("连接已关闭，无法发布")
        return
    }

    // --- JetStream 错误 ---

    // ErrStreamNotFound：Stream 不存在
    _, err = js.Stream(ctx, "NONEXISTENT")
    if errors.Is(err, jetstream.ErrStreamNotFound) {
        log.Println("Stream 不存在，可能需要先创建")
        return
    }

    // ErrConsumerNotFound：Consumer 不存在
    _, err = js.Consumer(ctx, "ORDERS", "nonexistent-consumer")
    if errors.Is(err, jetstream.ErrConsumerNotFound) {
        log.Println("Consumer 不存在")
        return
    }

    // ErrKeyNotFound：KV Key 不存在
    kv, _ := js.KeyValue(ctx, "APP_CONFIG")
    _, err = kv.Get(ctx, "missing.key")
    if errors.Is(err, jetstream.ErrKeyNotFound) {
        log.Println("Key 不存在，使用默认值")
        return
    }

    // ErrKeyExists：KV Create 时 Key 已存在
    _, err = kv.Create(ctx, "existing.key", []byte("value"))
    if errors.Is(err, jetstream.ErrKeyExists) {
        log.Println("Key 已存在（抢锁失败）")
        return
    }

    // ErrKeyWrongLastRevision：KV Update 时版本号不匹配
    _, err = kv.Update(ctx, "config.key", []byte("new"), 999)
    if errors.Is(err, jetstream.ErrKeyWrongLastRevision) {
        log.Println("并发修改冲突，需要重新读取并重试")
        return
    }

    // ErrNoMessages：Fetch 时没有可用消息
    consumer, _ := js.Consumer(ctx, "ORDERS", "my-consumer")
    msgs, _ := consumer.FetchNoWait(10)
    for msg := range msgs.Messages() {
        _ = msg
    }
    if errors.Is(msgs.Error(), jetstream.ErrNoMessages) {
        log.Println("当前没有待消费的消息")
    }

    // jetstream.APIError：JetStream API 返回的业务错误
    var apiErr *jetstream.APIError
    _, err = js.CreateStream(ctx, jetstream.StreamConfig{
        Name:     "EXISTING_STREAM",
        Subjects: []string{"other.subject"}, // Subject 与已有 Stream 冲突
    })
    if errors.As(err, &apiErr) {
        log.Printf("JetStream API 错误: code=%d description=%s",
            apiErr.ErrorCode, apiErr.Description)
        // ErrorCode 参考 NATS 文档
        // 10058 = Stream 名称已存在
        // 10065 = Subject 已被其他 Stream 使用
    }
}

5.2 连接断开时的处理

// 在发布时处理连接断开的最佳实践
func publishWithRetry(nc *nats.Conn, js jetstream.JetStream, subject string, data []byte) error {
    maxRetries := 3
    backoff := 500 * time.Millisecond

    for i := 0; i < maxRetries; i++ {
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

        // JetStream Publish：连接断开期间消息缓冲在内存中，重连后自动重发
        ack, err := js.Publish(ctx, subject, data)
        cancel()

        if err == nil {
            // 成功
            log.Printf("发布成功: stream=%s seq=%d", ack.Stream, ack.Sequence)
            return nil
        }

        // 判断是否应该重试
        if errors.Is(err, nats.ErrConnectionClosed) {
            // 连接已永久关闭，不应重试
            return fmt.Errorf("连接已关闭，放弃发布: %w", err)
        }

        if i < maxRetries-1 {
            log.Printf("发布失败，%v 后重试（第%d次）: %v", backoff, i+1, err)
            time.Sleep(backoff)
            backoff *= 2 // 指数退避
        }
    }

    return fmt.Errorf("发布失败，已重试 %d 次", maxRetries)
}

6. 性能优化

6.1 发布端优化：PublishAsync + AcksFutures

// 高吞吐发布模式：批量异步发布
func highThroughputPublish(js jetstream.JetStream) error {
    ctx := context.Background()

    const batchSize = 1000
    const totalMessages = 100000

    // 使用 PublishAsyncMaxPending 控制飞行中的消息数
    // 防止内存无限增长
    js, err := jetstream.New(nc,
        jetstream.WithPublishAsyncMaxPending(512),
    )

    var pendingAcks []jetstream.PubAckFuture
    startTime := time.Now()

    for i := 0; i < totalMessages; i++ {
        data := []byte(fmt.Sprintf(`{"seq":%d}`, i))

        // 异步发布，立即返回 Future
        paf, err := js.PublishAsync("orders.placed", data,
            // 设置消息 ID 用于去重
            jetstream.WithMsgID(fmt.Sprintf("msg-%d", i)),
        )
        if err != nil {
            return fmt.Errorf("PublishAsync 失败: %w", err)
        }
        pendingAcks = append(pendingAcks, paf)

        // 每 batchSize 条消息等待一批 ACK（控制内存和飞行消息数）
        if len(pendingAcks) >= batchSize {
            if err := waitForAcks(ctx, pendingAcks); err != nil {
                return err
            }
            pendingAcks = pendingAcks[:0] // 清空切片，复用底层数组
        }
    }

    // 等待最后一批
    if len(pendingAcks) > 0 {
        if err := waitForAcks(ctx, pendingAcks); err != nil {
            return err
        }
    }

    elapsed := time.Since(startTime)
    rate := float64(totalMessages) / elapsed.Seconds()
    fmt.Printf("发布完成: %d 条消息, 耗时 %v, %.0f msg/s\n",
        totalMessages, elapsed.Round(time.Millisecond), rate)
    return nil
}

func waitForAcks(ctx context.Context, pendingAcks []jetstream.PubAckFuture) error {
    for _, paf := range pendingAcks {
        select {
        case ack := <-paf.Ok():
            _ = ack // ACK 成功
        case err := <-paf.Err():
            return fmt.Errorf("PublishAsync ACK 失败: %w", err)
        case <-ctx.Done():
            return ctx.Err()
        }
    }
    return nil
}

6.2 消费端优化：批量 Fetch + Worker Pool

func workerPoolConsumer(js jetstream.JetStream, numWorkers, batchSize int) {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    consumer, err := js.Consumer(ctx, "ORDERS", "order-processor")
    if err != nil {
        log.Fatal(err)
    }

    // 消息通道（Worker Pool 缓冲）
    msgChan := make(chan jetstream.Msg, numWorkers*batchSize)

    // 启动 Worker Pool
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for msg := range msgChan {
                if err := processOrder(msg); err != nil {
                    log.Printf("[worker-%d] 处理失败，NAK: %v", workerID, err)
                    msg.NakWithDelay(5 * time.Second)
                } else {
                    msg.Ack()
                }
            }
        }(i)
    }

    // 主循环：批量 Fetch 并分发给 Worker
    ticker := time.NewTicker(100 * time.Millisecond) // 每 100ms 拉一批
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            close(msgChan)
            wg.Wait()
            return

        case <-ticker.C:
            msgs, err := consumer.FetchNoWait(batchSize)
            if err != nil {
                log.Printf("Fetch 失败: %v", err)
                continue
            }

            count := 0
            for msg := range msgs.Messages() {
                select {
                case msgChan <- msg:
                    count++
                case <-ctx.Done():
                    msg.Nak() // 不处理，归还给队列
                    return
                }
            }

            if count > 0 {
                log.Printf("本批次获取 %d 条消息", count)
            }
        }
    }
}

func processOrder(msg jetstream.Msg) error {
    // 业务处理逻辑
    fmt.Printf("处理订单: %s\n", string(msg.Data()))
    time.Sleep(10 * time.Millisecond) // 模拟处理耗时
    return nil
}

6.3 连接复用原则

// 最佳实践：整个应用共用一个 NATS 连接
// NATS 连接内部使用多路复用（multiplexing），无需为不同业务创建多个连接

// --- 推荐方式 ---
type App struct {
    nc *nats.Conn
    js jetstream.JetStream
}

func NewApp(natsURL string) (*App, error) {
    nc, err := nats.Connect(natsURL,
        nats.Name("my-app"),
        nats.MaxReconnects(-1),
    )
    if err != nil {
        return nil, err
    }

    js, err := jetstream.New(nc)
    if err != nil {
        return nil, err
    }

    return &App{nc: nc, js: js}, nil
}

// 不同服务层共享同一个连接
func (a *App) OrderService() *OrderService {
    return &OrderService{js: a.js}
}

func (a *App) NotificationService() *NotificationService {
    return &NotificationService{nc: a.nc}
}

// --- 不推荐的方式（资源浪费）---
// 每个请求或每个服务都创建新连接，会导致：
// - 连接数暴涨（NATS Server 默认限制 65536 个连接）
// - 每个连接都有独立的 goroutine 和内存开销
// - 更多的 TCP 握手和 TLS 开销
func badPractice(natsURL string) {
    for i := 0; i < 100; i++ {
        nc, _ := nats.Connect(natsURL) // 不推荐！每次操作创建新连接
        nc.Publish("test", []byte("data"))
        nc.Close()
    }
}

// 性能参考数据（单节点，普通硬件）：
// 单连接 Core NATS Publish:    ~2,000,000 msg/s
// 单连接 JetStream Publish (Sync): ~30,000 msg/s（受 ACK 往返延迟限制）
// 单连接 JetStream Publish (Async): ~200,000 msg/s（批量等 ACK）
// 多个连接主要在 subscribe fanout 场景有少量提升，发布场景基本无帮助

7. 完整实际示例

7.1 订单事件生产者

// producer/main.go
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "math/rand"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

type Order struct {
    ID        string    `json:"id"`
    UserID    string    `json:"user_id"`
    Amount    float64   `json:"amount"`
    Status    string    `json:"status"`
    CreatedAt time.Time `json:"created_at"`
}

type OrderProducer struct {
    js        jetstream.JetStream
    streamName string
}

func NewOrderProducer(nc *nats.Conn) (*OrderProducer, error) {
    js, err := jetstream.New(nc)
    if err != nil {
        return nil, fmt.Errorf("创建 JetStream context 失败: %w", err)
    }

    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    // 确保 Stream 存在（幂等操作）
    _, err = js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
        Name:        "ORDERS",
        Description: "订单事件流",
        Subjects:    []string{"orders.>"},
        Replicas:    1, // 开发环境用 1，生产环境用 3
        Storage:     jetstream.FileStorage,
        Retention:   jetstream.LimitsPolicy,
        Discard:     jetstream.DiscardOld,
        MaxAge:      7 * 24 * time.Hour,
        MaxBytes:    1 << 30, // 1GB
        Duplicates:  5 * time.Minute, // 5分钟去重窗口
    })
    if err != nil {
        return nil, fmt.Errorf("创建 Stream 失败: %w", err)
    }

    return &OrderProducer{js: js, streamName: "ORDERS"}, nil
}

func (p *OrderProducer) PlaceOrder(ctx context.Context, order Order) error {
    data, err := json.Marshal(order)
    if err != nil {
        return fmt.Errorf("序列化失败: %w", err)
    }

    // Subject 格式：orders.placed
    // 使用 Nats-Msg-Id 消息头实现幂等发布（防止网络重试导致重复）
    msg := &nats.Msg{
        Subject: "orders.placed",
        Data:    data,
        Header: nats.Header{
            "Nats-Msg-Id":  []string{order.ID}, // 去重 ID
            "Content-Type": []string{"application/json"},
        },
    }

    pubCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    ack, err := p.js.PublishMsg(pubCtx, msg)
    if err != nil {
        return fmt.Errorf("发布失败: %w", err)
    }

    if ack.Duplicate {
        log.Printf("[producer] 订单 %s 已存在（幂等处理，忽略）", order.ID)
        return nil
    }

    log.Printf("[producer] 订单 %s 已发布: stream=%s seq=%d",
        order.ID, ack.Stream, ack.Sequence)
    return nil
}

func main() {
    natsURL := os.Getenv("NATS_URL")
    if natsURL == "" {
        natsURL = "nats://localhost:4222"
    }

    nc, err := nats.Connect(natsURL,
        nats.Name("order-producer"),
        nats.MaxReconnects(-1),
        nats.ReconnectWait(2*time.Second),
        nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
            log.Printf("[producer] 连接断开: %v", err)
        }),
        nats.ReconnectHandler(func(nc *nats.Conn) {
            log.Printf("[producer] 已重连: %s", nc.ConnectedUrl())
        }),
    )
    if err != nil {
        log.Fatal("连接 NATS 失败:", err)
    }
    defer nc.Drain()

    producer, err := NewOrderProducer(nc)
    if err != nil {
        log.Fatal("初始化生产者失败:", err)
    }

    // 模拟持续产生订单
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, os.Interrupt, syscall.SIGTERM)

    ticker := time.NewTicker(500 * time.Millisecond)
    defer ticker.Stop()

    orderSeq := 0
    ctx := context.Background()

    for {
        select {
        case <-quit:
            log.Println("[producer] 收到退出信号，停止生产者")
            return

        case <-ticker.C:
            orderSeq++
            order := Order{
                ID:        fmt.Sprintf("ORD-%06d", orderSeq),
                UserID:    fmt.Sprintf("USR-%04d", rand.Intn(1000)),
                Amount:    float64(rand.Intn(10000)) / 100.0,
                Status:    "pending",
                CreatedAt: time.Now(),
            }

            if err := producer.PlaceOrder(ctx, order); err != nil {
                log.Printf("[producer] 发布订单失败: %v", err)
            }
        }
    }
}

7.2 订单事件消费者（Pull Consumer + Worker Pool）

// consumer/main.go
package main

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "log"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"

    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

type Order struct {
    ID        string    `json:"id"`
    UserID    string    `json:"user_id"`
    Amount    float64   `json:"amount"`
    Status    string    `json:"status"`
    CreatedAt time.Time `json:"created_at"`
}

type OrderConsumer struct {
    js       jetstream.JetStream
    consumer jetstream.Consumer
}

func NewOrderConsumer(nc *nats.Conn, consumerName string) (*OrderConsumer, error) {
    js, err := jetstream.New(nc)
    if err != nil {
        return nil, err
    }

    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    // 创建或更新 Consumer（幂等）
    cons, err := js.CreateOrUpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
        Name:          consumerName,
        Durable:       consumerName,
        Description:   "订单处理消费者",
        FilterSubject: "orders.placed",
        DeliverPolicy: jetstream.DeliverAllPolicy,
        AckPolicy:     jetstream.AckExplicitPolicy,
        AckWait:       60 * time.Second, // 给业务处理足够时间
        MaxDeliver:    5,                // 最多重投 5 次
        MaxAckPending: 500,             // 限制并发，防止内存暴涨
        BackOff: []time.Duration{
            5 * time.Second,
            15 * time.Second,
            60 * time.Second,
            5 * time.Minute,
        },
    })
    if err != nil {
        return nil, fmt.Errorf("创建 Consumer 失败: %w", err)
    }

    return &OrderConsumer{js: js, consumer: cons}, nil
}

// 处理单个订单
func processOrder(order Order) error {
    log.Printf("[worker] 处理订单: id=%s user=%s amount=%.2f",
        order.ID, order.UserID, order.Amount)

    // 模拟处理耗时（真实场景中这里调用数据库、支付API等）
    time.Sleep(50 * time.Millisecond)

    // 模拟偶发失败（5% 概率）
    if len(order.ID)%20 == 0 {
        return fmt.Errorf("模拟处理失败: %s", order.ID)
    }

    return nil
}

// Start 启动消费者（带 Worker Pool）
func (c *OrderConsumer) Start(ctx context.Context, numWorkers int) {
    msgChan := make(chan jetstream.Msg, numWorkers*10)

    // 启动 Worker Pool
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            log.Printf("[worker-%d] 已启动", workerID)

            for {
                select {
                case msg, ok := <-msgChan:
                    if !ok {
                        log.Printf("[worker-%d] 通道关闭，退出", workerID)
                        return
                    }

                    // 解析消息
                    var order Order
                    if err := json.Unmarshal(msg.Data(), &order); err != nil {
                        log.Printf("[worker-%d] 消息解析失败，Term: %v", workerID, err)
                        msg.Term() // 无效消息，永久丢弃
                        continue
                    }

                    // 处理订单
                    if err := processOrder(order); err != nil {
                        meta, _ := msg.Metadata()
                        log.Printf("[worker-%d] 处理失败 (第%d次): %v",
                            workerID, meta.NumDelivered, err)

                        if meta.NumDelivered >= 5 {
                            // 已重试 5 次，永久标记失败
                            log.Printf("[worker-%d] 订单 %s 超过最大重试次数，Term",
                                workerID, order.ID)
                            msg.Term()
                        } else {
                            // 延迟重试
                            msg.NakWithDelay(10 * time.Second)
                        }
                    } else {
                        msg.Ack()
                        log.Printf("[worker-%d] 订单 %s 处理成功", workerID, order.ID)
                    }

                case <-ctx.Done():
                    return
                }
            }
        }(i)
    }

    // 主 Fetch 循环
    go func() {
        defer func() {
            close(msgChan)
            wg.Wait()
            log.Println("[consumer] 所有 Worker 已停止")
        }()

        iter, err := c.consumer.Messages(
            jetstream.PullMaxMessages(100),
            jetstream.PullHeartbeat(10*time.Second),
        )
        if err != nil {
            log.Printf("[consumer] 启动消息迭代器失败: %v", err)
            return
        }
        defer iter.Stop()

        for {
            msg, err := iter.Next()
            if err != nil {
                if errors.Is(err, jetstream.ErrMsgIteratorClosed) {
                    log.Println("[consumer] 消息迭代器已关闭")
                    return
                }
                if ctx.Err() != nil {
                    log.Println("[consumer] Context 取消，停止消费")
                    return
                }
                log.Printf("[consumer] iter.Next() 错误: %v", err)
                time.Sleep(1 * time.Second) // 短暂等待后重试
                continue
            }

            select {
            case msgChan <- msg:
                // 消息已分发给 Worker
            case <-ctx.Done():
                msg.Nak() // 归还消息
                return
            }
        }
    }()
}

func main() {
    natsURL := os.Getenv("NATS_URL")
    if natsURL == "" {
        natsURL = "nats://localhost:4222"
    }

    consumerName := os.Getenv("CONSUMER_NAME")
    if consumerName == "" {
        consumerName = "order-processor"
    }

    nc, err := nats.Connect(natsURL,
        nats.Name("order-consumer"),
        nats.MaxReconnects(-1),
    )
    if err != nil {
        log.Fatal("连接 NATS 失败:", err)
    }
    defer nc.Drain()

    consumer, err := NewOrderConsumer(nc, consumerName)
    if err != nil {
        log.Fatal("初始化消费者失败:", err)
    }

    ctx, cancel := context.WithCancel(context.Background())

    // 优雅停机
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, os.Interrupt, syscall.SIGTERM)

    go func() {
        <-quit
        log.Println("[consumer] 收到退出信号，开始优雅停机...")
        cancel()
    }()

    log.Printf("[consumer] 启动消费者（4个 Worker）")
    consumer.Start(ctx, 4)

    <-ctx.Done()
    log.Println("[consumer] 消费者已停止")
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225

7.3 Request/Reply 微服务

// microservice/main.go
// 演示基于 NATS Request/Reply 的微服务通信
package main

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/nats-io/nats.go"
)

// --- 数据结构 ---

type UserRequest struct {
    UserID string `json:"user_id"`
}

type UserResponse struct {
    UserID string `json:"user_id"`
    Name   string `json:"name"`
    Email  string `json:"email"`
    Error  string `json:"error,omitempty"`
}

type InventoryRequest struct {
    ProductID string `json:"product_id"`
    Quantity  int    `json:"quantity"`
}

type InventoryResponse struct {
    Available bool   `json:"available"`
    Stock     int    `json:"stock"`
    Error     string `json:"error,omitempty"`
}

// --- User Service ---

type UserService struct {
    nc *nats.Conn
}

func NewUserService(nc *nats.Conn) *UserService {
    return &UserService{nc: nc}
}

func (s *UserService) Start() error {
    // 注册处理函数
    _, err := s.nc.Subscribe("rpc.user.get", s.handleGetUser)
    if err != nil {
        return err
    }

    // 使用 QueueSubscribe 实现水平扩展（多个 User Service 实例共享负载）
    _, err = s.nc.QueueSubscribe("rpc.user.validate", "user-service", s.handleValidateUser)
    return err
}

func (s *UserService) handleGetUser(msg *nats.Msg) {
    var req UserRequest
    if err := json.Unmarshal(msg.Data, &req); err != nil {
        resp := UserResponse{Error: "请求格式错误"}
        data, _ := json.Marshal(resp)
        msg.Respond(data)
        return
    }

    // 模拟数据库查询
    resp := UserResponse{
        UserID: req.UserID,
        Name:   fmt.Sprintf("用户_%s", req.UserID),
        Email:  fmt.Sprintf("%s@example.com", req.UserID),
    }

    data, _ := json.Marshal(resp)
    if err := msg.Respond(data); err != nil {
        log.Printf("[user-service] Respond 失败: %v", err)
    }
}

func (s *UserService) handleValidateUser(msg *nats.Msg) {
    // 验证用户逻辑
    msg.Respond([]byte(`{"valid": true}`))
}

// --- Inventory Service ---

type InventoryService struct {
    nc *nats.Conn
}

func NewInventoryService(nc *nats.Conn) *InventoryService {
    return &InventoryService{nc: nc}
}

func (s *InventoryService) Start() error {
    _, err := s.nc.QueueSubscribe("rpc.inventory.check", "inventory-service",
        func(msg *nats.Msg) {
            var req InventoryRequest
            if err := json.Unmarshal(msg.Data, &req); err != nil {
                resp := InventoryResponse{Error: "请求格式错误"}
                data, _ := json.Marshal(resp)
                msg.Respond(data)
                return
            }

            // 模拟库存查询
            stock := 100
            resp := InventoryResponse{
                Available: stock >= req.Quantity,
                Stock:     stock,
            }

            data, _ := json.Marshal(resp)
            msg.Respond(data)
        })
    return err
}

// --- API Gateway（聚合多个微服务）---

type APIGateway struct {
    nc *nats.Conn
}

func NewAPIGateway(nc *nats.Conn) *APIGateway {
    return &APIGateway{nc: nc}
}

// NATSRequest 封装 NATS Request/Reply，带错误处理
func (g *APIGateway) NATSRequest(ctx context.Context, subject string, payload interface{}, response interface{}) error {
    data, err := json.Marshal(payload)
    if err != nil {
        return fmt.Errorf("序列化请求失败: %w", err)
    }

    // 从 Context 获取超时
    deadline, ok := ctx.Deadline()
    timeout := 5 * time.Second
    if ok {
        timeout = time.Until(deadline)
    }

    msg, err := g.nc.Request(subject, data, timeout)
    if err != nil {
        if errors.Is(err, nats.ErrNoResponders) {
            return fmt.Errorf("服务 %s 不可用", subject)
        }
        if errors.Is(err, nats.ErrTimeout) {
            return fmt.Errorf("服务 %s 响应超时", subject)
        }
        return fmt.Errorf("请求服务 %s 失败: %w", subject, err)
    }

    if err := json.Unmarshal(msg.Data, response); err != nil {
        return fmt.Errorf("解析响应失败: %w", err)
    }
    return nil
}

// HandlePlaceOrder HTTP handler 聚合多个微服务
func (g *APIGateway) HandlePlaceOrder(w http.ResponseWriter, r *http.Request) {
    ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
    defer cancel()

    userID := r.URL.Query().Get("user_id")
    productID := r.URL.Query().Get("product_id")

    // 并发请求 User Service 和 Inventory Service
    type result struct {
        userResp UserResponse
        invResp  InventoryResponse
        err      error
    }

    resultChan := make(chan result, 1)

    go func() {
        var r result

        // 1. 验证用户
        if err := g.NATSRequest(ctx, "rpc.user.get",
            UserRequest{UserID: userID}, &r.userResp); err != nil {
            r.err = fmt.Errorf("获取用户失败: %w", err)
            resultChan <- r
            return
        }
        if r.userResp.Error != "" {
            r.err = fmt.Errorf("用户错误: %s", r.userResp.Error)
            resultChan <- r
            return
        }

        // 2. 检查库存
        if err := g.NATSRequest(ctx, "rpc.inventory.check",
            InventoryRequest{ProductID: productID, Quantity: 1}, &r.invResp); err != nil {
            r.err = fmt.Errorf("检查库存失败: %w", err)
            resultChan <- r
            return
        }

        resultChan <- r
    }()

    res := <-resultChan
    if res.err != nil {
        http.Error(w, res.err.Error(), http.StatusInternalServerError)
        return
    }

    if !res.invResp.Available {
        http.Error(w, "库存不足", http.StatusConflict)
        return
    }

    // 3. 创建订单（这里简化，实际应写入数据库并发布 JetStream 事件）
    response := map[string]interface{}{
        "order_id":   fmt.Sprintf("ORD-%d", time.Now().UnixNano()),
        "user":       res.userResp,
        "product_id": productID,
        "status":     "created",
    }

    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(response)
}

func main() {
    natsURL := os.Getenv("NATS_URL")
    if natsURL == "" {
        natsURL = "nats://localhost:4222"
    }

    nc, err := nats.Connect(natsURL,
        nats.Name("rpc-demo"),
        nats.MaxReconnects(-1),
    )
    if err != nil {
        log.Fatal("连接 NATS 失败:", err)
    }
    defer nc.Drain()

    // 启动各微服务
    userSvc := NewUserService(nc)
    if err := userSvc.Start(); err != nil {
        log.Fatal("启动 User Service 失败:", err)
    }

    invSvc := NewInventoryService(nc)
    if err := invSvc.Start(); err != nil {
        log.Fatal("启动 Inventory Service 失败:", err)
    }

    // 启动 API Gateway
    gateway := NewAPIGateway(nc)
    mux := http.NewServeMux()
    mux.HandleFunc("/api/orders", gateway.HandlePlaceOrder)

    srv := &http.Server{
        Addr:    ":8080",
        Handler: mux,
    }

    // 优雅停机
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, os.Interrupt, syscall.SIGTERM)

    go func() {
        log.Println("API Gateway 已启动在 :8080")
        log.Println("测试: curl 'http://localhost:8080/api/orders?user_id=U001&product_id=P001'")
        if err := srv.ListenAndServe(); err != http.ErrServerClosed {
            log.Fatal(err)
        }
    }()

    <-quit
    log.Println("收到退出信号，开始优雅停机...")

    shutCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    srv.Shutdown(shutCtx)
    log.Println("服务已停止")
}

总结

场景	推荐 API	关键注意点
实时事件广播	`nc.Publish` + `nc.Subscribe`	无持久化，消费者离线会丢消息
负载均衡消费	`nc.QueueSubscribe`	Core NATS 级别，无 ACK
微服务 RPC	`nc.Request` / `nc.Subscribe` + `msg.Respond`	处理 ErrNoResponders 和 ErrTimeout
事件持久化	`js.Publish` + Pull Consumer	等待 PubAck 确认持久化
高吞吐发布	`js.PublishAsync`	批量等待 AcksFutures
可靠消费	Pull Consumer + Worker Pool	显式 ACK，处理失败用 NakWithDelay
配置热更新	KV Store + Watch	初始值 + 增量推送
连接管理	全局单连接 + 回调	MaxReconnects=-1，使用 Drain 优雅关闭

Go 客户端实战指南 ​

开发环境连接信息 ​

目录 ​

1. 依赖安装 ​

2. 连接管理 ​

2.1 基础连接 ​

2.2 连接选项 ​

2.3 连接事件回调 ​

2.4 多节点连接（集群高可用） ​

2.5 TLS 连接 ​

2.6 drain vs close 的区别 ​

3. Core NATS 使用 ​

3.1 Publish — 发布 ​

3.2 Subscribe — 订阅 ​

3.3 QueueSubscribe — 负载均衡消费 ​

3.4 Request/Reply — 请求响应 ​

4. JetStream v2 API ​

4.1 获取 JetStream Context ​

4.2 Stream 操作 ​

4.3 Publish — 发布消息到 Stream ​

4.4 Consumer 操作与 Pull Consumer 消费 ​

4.5 Push Consumer 消费 ​

4.6 KV Store 操作（见第 07 篇完整示例） ​

5. 错误处理模式 ​

5.1 常见错误类型 ​

5.2 连接断开时的处理 ​

6. 性能优化 ​

6.1 发布端优化：PublishAsync + AcksFutures ​

6.2 消费端优化：批量 Fetch + Worker Pool ​

6.3 连接复用原则 ​

7. 完整实际示例 ​

7.1 订单事件生产者 ​

7.2 订单事件消费者（Pull Consumer + Worker Pool） ​

7.3 Request/Reply 微服务 ​

总结 ​