从零开始系统学习 Go 编程

Appearance

KV Store 与 Object Store ​

目录 ​

  1. KV Store 概念
  2. 内部实现原理
  3. KV Store vs Redis 对比
  4. KeyValue Config 详解
  5. 操作 API 全解
  6. Watch 机制详解
  7. History 版本历史
  8. 分布式锁实现模式
  9. 配置中心模式
  10. Object Store 概念
  11. Object Store vs KV Store 区别
  12. Object Store 常用操作
  13. 完整代码示例

1. KV Store 概念 ​

NATS JetStream KV Store（键值存储）是 NATS 2.x 引入的一种高级抽象，它在 JetStream Stream 之上提供了类似 etcd/Consul 的键值语义，原生支持：

  • 持久化存储：数据写入磁盘，服务重启后不丢失
  • 版本历史：每次 Put 都产生新的 Revision，支持回溯任意历史版本
  • TTL 过期：Key 可以设置自动过期时间
  • Watch 推送：客户端可以订阅 key 变化，实现实时推送（无需轮询）
  • CAS 操作：Create / Update 支持条件写入，天然适合分布式锁和领导选举
  • 高可用副本：底层 Stream 支持多副本，天然支持集群容错

KV Store 的核心使用场景：

场景说明
配置中心存储服务配置，Watch 实现热更新
服务发现注册服务实例，TTL 心跳续期
分布式锁利用 Create CAS 实现互斥锁
特征开关Feature Flag，实时下发开关
小型元数据存储记录任务状态、调度元数据

注意：KV Store 设计用于存储相对小的 Value（建议单个 Value < 1MB）。对于大文件存储，应使用 Object Store。

2. 内部实现原理 ​

KV Store 本质上是对 JetStream Stream 的一层语义封装，理解底层实现有助于排查问题和做容量规划。

2.1 底层 Stream 命名规则 ​

创建一个名为 MYCONFIG 的 KV Bucket，NATS 会在内部自动创建一个 Stream：

Stream 名称：KV_MYCONFIG
Subject 过滤：$KV.MYCONFIG.>

可以通过 NATS CLI 或监控端点直接看到这个 Stream：

bash
nats stream ls
# 输出中会看到 KV_MYCONFIG

nats stream info KV_MYCONFIG

2.2 Subject 路由规则 ​

每个 KV 操作都映射到特定的 Subject：

Put("app.version", "1.2.3")  -->  Subject: $KV.MYCONFIG.app.version
Put("db.host", "localhost")   -->  Subject: $KV.MYCONFIG.db.host
Get("app.version")            -->  直接读取 Stream 的最新消息

Key 中的 . 是合法字符，它在 NATS Subject 中自然成为层级分隔符，可以用通配符 Watch。

2.3 消息头字段（KV 操作类型） ​

不同的 KV 操作通过消息头 KV-Operation 区分：

操作KV-Operation 头说明
Put（无）普通写入
DeleteDEL软删除，写入一条带删除标记的消息
PurgePURGE清除历史，写入清除标记

这意味着 Delete 并非真正删除数据，而是追加一条删除标记消息。旧的历史记录在 History 窗口内仍然可以查询。

2.4 Stream 配置映射 ​

KV Config 与底层 Stream Config 的对应关系：

KV Bucket Config         -->  Stream Config
──────────────────────────────────────────────
Bucket     "MYCONFIG"    -->  Name       "KV_MYCONFIG"
History    5             -->  MaxMsgsPerSubject 5
TTL        24h           -->  MaxAge     24h
MaxValueSize 1MB         -->  MaxMsgSize 1MB
Replicas   3             -->  Replicas   3
Storage    File          -->  Storage    File

2.5 Revision 与 Stream Sequence 的关系 ​

KV 的 Revision 号直接等于底层 Stream 的 Sequence 号，这是全局单调递增的。因此 Revision 可以跨 Key 比较先后顺序。

Put("a", "1")  -->  Revision=1, Sequence=1
Put("b", "2")  -->  Revision=2, Sequence=2
Put("a", "2")  -->  Revision=3, Sequence=3

3. KV Store vs Redis 对比 ​

KV Store 不是 Redis 的替代品，二者定位不同，各有侧重：

特性NATS KV StoreRedis
定位消息系统内置的元数据存储通用高性能缓存/数据库
写入性能中等（JetStream 持久化路径）极高（内存操作）
读取性能中等极高
数据结构仅 KV（String value）丰富（Hash/List/Set/ZSet/Stream）
Watch 推送原生支持，基于 NATS 订阅需要 Keyspace Notification（有限制）
版本历史原生支持（History 参数）不支持
TTL支持（Bucket 级别）支持（Key 级别，更灵活）
CAS 操作Create/Update 原生支持需要 WATCH + MULTI/EXEC
持久化原生（Write-ahead log）可选（RDB/AOF）
集群副本原生 Raft 共识Redis Cluster/Sentinel
运维复杂度低（NATS 一套系统）需要单独部署维护
适用 Value 大小< 1MB（推荐）无严格限制（但建议 < 100KB）

什么时候用 NATS KV Store？

  • 你的系统已经使用 NATS，希望减少外部依赖
  • 需要 Watch 推送能力（配置热更新、Feature Flag）
  • 需要版本历史和回溯能力
  • 需要简单的分布式锁
  • 数据量不大，对写入延迟不敏感（< 10ms 级别即可）

什么时候坚持使用 Redis？

  • 需要极低延迟（微秒级）缓存
  • 需要 Hash/List/ZSet 等复杂数据结构
  • 需要 Key 级别的独立 TTL
  • 需要 Lua 脚本原子操作
  • 数据量大，需要内存数据库的吞吐能力

4. KeyValue Config 详解 ​

go
type KeyValueConfig struct {
    // Bucket 名称（必填），等同于 KV 命名空间
    // 底层 Stream 名为 KV_<Bucket>
    Bucket string

    // 描述信息（可选）
    Description string

    // TTL：Key 的最大存活时间
    // 设置后，超过此时间未更新的 Key 会被自动删除
    // 对应底层 Stream 的 MaxAge
    // 0 表示不过期
    TTL time.Duration

    // Storage：存储类型
    // nats.FileStorage（默认，持久化到磁盘）
    // nats.MemoryStorage（仅内存，重启丢失）
    Storage StorageType

    // Replicas：副本数
    // 1=无副本（默认，开发环境）
    // 3=3副本（生产环境推荐，可容忍1节点故障）
    // 必须 <= 集群节点数
    Replicas int

    // MaxValueSize：单个 Value 的最大字节数
    // -1 表示不限制
    // 推荐设置，防止误写超大 Value
    MaxValueSize int32

    // History：保留每个 Key 的历史版本数
    // 1=只保留最新值（默认）
    // 最大 64
    History uint8

    // MaxBytes：整个 Bucket 的最大存储字节数
    // -1 表示不限制
    MaxBytes int64

    // MaxChunkSize：与 ObjectStore 相关，KV 忽略此项
    // （内部使用）
}

4.1 典型配置示例 ​

go
// 开发环境：内存存储，无副本，无过期
devConfig := jetstream.KeyValueConfig{
    Bucket:       "DEV_CONFIG",
    Storage:      jetstream.MemoryStorage,
    Replicas:     1,
    History:      5,
}

// 生产环境：文件存储，3副本，30天TTL
prodConfig := jetstream.KeyValueConfig{
    Bucket:       "PROD_CONFIG",
    Description:  "生产环境服务配置",
    Storage:      jetstream.FileStorage,
    Replicas:     3,
    History:      10,      // 保留10个历史版本
    TTL:          30 * 24 * time.Hour,
    MaxValueSize: 1 << 20, // 1MB
    MaxBytes:     1 << 30, // 1GB 总上限
}

// 服务注册：短 TTL，用于心跳感知服务存活
registryConfig := jetstream.KeyValueConfig{
    Bucket:   "SERVICE_REGISTRY",
    TTL:      30 * time.Second, // 30秒无心跳则自动过期
    Replicas: 3,
    History:  1, // 只需最新状态
}

5. 操作 API 全解 ​

5.1 初始化 ​

go
package main

import (
    "context"
    "fmt"
    "log"
    "time"

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

func initKV() (jetstream.KeyValue, func()) {
    // 建立连接
    nc, err := nats.Connect("nats://localhost:4222")
    if err != nil {
        log.Fatal(err)
    }

    // 创建 JetStream context
    js, err := jetstream.New(nc)
    if err != nil {
        log.Fatal(err)
    }

    ctx := context.Background()

    // 创建或更新 KV Bucket
    // CreateKeyValue：若已存在则返回错误
    // UpdateKeyValue：更新已有 Bucket 配置
    // CreateOrUpdateKeyValue：推荐，幂等操作
    kv, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
        Bucket:       "MYAPP",
        Description:  "应用配置存储",
        Replicas:     1,
        History:      5,
        TTL:          24 * time.Hour,
        MaxValueSize: 512 * 1024, // 512KB
    })
    if err != nil {
        log.Fatal(err)
    }

    cleanup := func() { nc.Drain() }
    return kv, cleanup
}

5.2 Put — 写入/更新 ​

go
func examplePut(kv jetstream.KeyValue) {
    ctx := context.Background()

    // 基础写入：返回新的 Revision 号
    rev, err := kv.Put(ctx, "database.host", []byte("127.0.0.1"))
    if err != nil {
        log.Printf("Put 失败: %v", err)
        return
    }
    fmt.Printf("写入成功，Revision: %d\n", rev) // Revision: 1

    // 再次 Put 同一个 key，Revision 递增
    rev, err = kv.Put(ctx, "database.host", []byte("10.0.0.1"))
    if err != nil {
        log.Printf("Put 失败: %v", err)
        return
    }
    fmt.Printf("更新成功，Revision: %d\n", rev) // Revision: 2

    // 写入 JSON 数据
    type DBConfig struct {
        Host string `json:"host"`
        Port int    `json:"port"`
    }
    cfg := DBConfig{Host: "10.0.0.1", Port: 5432}
    data, _ := json.Marshal(cfg)
    rev, err = kv.Put(ctx, "database.config", data)
    fmt.Printf("写入 JSON，Revision: %d\n", rev)
}

5.3 Get — 读取 ​

go
func exampleGet(kv jetstream.KeyValue) {
    ctx := context.Background()

    // 读取最新值
    entry, err := kv.Get(ctx, "database.host")
    if err != nil {
        if errors.Is(err, jetstream.ErrKeyNotFound) {
            fmt.Println("Key 不存在")
            return
        }
        log.Printf("Get 失败: %v", err)
        return
    }

    // KeyValueEntry 包含完整元数据
    fmt.Printf("Key:       %s\n", entry.Key())
    fmt.Printf("Value:     %s\n", string(entry.Value()))
    fmt.Printf("Revision:  %d\n", entry.Revision())   // 全局版本号
    fmt.Printf("Delta:     %d\n", entry.Delta())       // 距最新版本的差值（Watch中有意义）
    fmt.Printf("Created:   %v\n", entry.Created())     // 创建时间
    fmt.Printf("Operation: %v\n", entry.Operation())   // Put/Delete/Purge
    fmt.Printf("Bucket:    %s\n", entry.Bucket())      // Bucket 名称

    // 读取指定历史版本（Revision）
    // 注意：只有在 History > 1 时才能读到历史版本
    oldEntry, err := kv.GetRevision(ctx, "database.host", 1)
    if err != nil {
        log.Printf("GetRevision 失败: %v", err)
        return
    }
    fmt.Printf("历史版本 1 的值: %s\n", string(oldEntry.Value()))
}

5.4 Create — 仅当 Key 不存在时写入（CAS） ​

go
func exampleCreate(kv jetstream.KeyValue) {
    ctx := context.Background()

    // Create 是原子 CAS 操作：
    // 如果 key 不存在 → 写入成功，返回 Revision
    // 如果 key 已存在 → 返回 ErrKeyExists 错误
    rev, err := kv.Create(ctx, "lock.job-processor", []byte("node-1"))
    if err != nil {
        if errors.Is(err, jetstream.ErrKeyExists) {
            fmt.Println("Key 已存在，Create 失败（正常，用于抢锁场景）")
            return
        }
        log.Printf("Create 失败: %v", err)
        return
    }
    fmt.Printf("Create 成功，Revision: %d\n", rev)
}

5.5 Update — 条件更新（版本号 CAS） ​

go
func exampleUpdate(kv jetstream.KeyValue) {
    ctx := context.Background()

    // 先读取当前值，获取 Revision
    entry, err := kv.Get(ctx, "app.config")
    if err != nil {
        log.Printf("Get 失败: %v", err)
        return
    }
    currentRev := entry.Revision()

    // Update：只有当当前 Revision == lastRev 时才写入成功
    // 如果期间有其他客户端修改了这个 key，Revision 会变化，Update 返回错误
    // 这实现了乐观锁语义
    newRev, err := kv.Update(ctx, "app.config", []byte("new-value"), currentRev)
    if err != nil {
        if errors.Is(err, jetstream.ErrKeyWrongLastRevision) {
            fmt.Println("版本号不匹配，有并发修改，需要重试")
            return
        }
        log.Printf("Update 失败: %v", err)
        return
    }
    fmt.Printf("Update 成功，新 Revision: %d\n", newRev)
}

5.6 Delete — 软删除 ​

go
func exampleDelete(kv jetstream.KeyValue) {
    ctx := context.Background()

    // Delete 是软删除：
    // 在底层 Stream 中写入一条带 KV-Operation: DEL 头的消息
    // Get 该 key 后会返回 ErrKeyNotFound
    // 但 History 中仍保留删除前的历史版本
    err := kv.Delete(ctx, "database.host")
    if err != nil {
        log.Printf("Delete 失败: %v", err)
        return
    }
    fmt.Println("Delete 成功（软删除）")

    // 验证：Get 返回 ErrKeyNotFound
    _, err = kv.Get(ctx, "database.host")
    if errors.Is(err, jetstream.ErrKeyNotFound) {
        fmt.Println("Get 返回 KeyNotFound，符合预期")
    }

    // 但 History 中仍可查到删除操作记录（如果 History > 1）
}

5.7 Purge — 清除所有历史 ​

go
func examplePurge(kv jetstream.KeyValue) {
    ctx := context.Background()

    // Purge 清除指定 key 的所有历史消息，只保留一条 PURGE 标记
    // 用于释放存储空间
    err := kv.Purge(ctx, "old.config.key")
    if err != nil {
        log.Printf("Purge 失败: %v", err)
        return
    }
    fmt.Println("Purge 成功，历史已清除")
}

5.8 Keys — 列出所有 Key ​

go
func exampleKeys(kv jetstream.KeyValue) {
    ctx := context.Background()

    // Keys 返回 Bucket 中所有当前存活的 Key（已删除/Purge 的不返回）
    keys, err := kv.Keys(ctx)
    if err != nil {
        if errors.Is(err, jetstream.ErrNoKeysFound) {
            fmt.Println("Bucket 为空")
            return
        }
        log.Printf("Keys 失败: %v", err)
        return
    }

    fmt.Printf("共 %d 个 Key:\n", len(keys))
    for _, key := range keys {
        fmt.Printf("  - %s\n", key)
    }

    // 带通配符过滤（仅列出 database. 前缀的 key）
    keys, err = kv.Keys(ctx, jetstream.IgnoreDeletes(),
        jetstream.MetaOnly()) // MetaOnly 只获取 key 名，不拉取 value，更高效
    if err != nil {
        log.Printf("Keys 失败: %v", err)
        return
    }
    fmt.Printf("过滤后 %d 个 Key\n", len(keys))
}

6. Watch 机制详解 ​

Watch 是 KV Store 最强大的特性之一。它基于 NATS 订阅实现，客户端不需要轮询，服务端主动推送变化。

6.1 Watch 工作原理 ​

                    KV Bucket (底层 Stream: KV_MYAPP)
                    ┌─────────────────────────────────┐
                    │  Seq=1: $KV.MYAPP.db.host = "a" │
                    │  Seq=2: $KV.MYAPP.db.port = "5" │
                    │  Seq=3: $KV.MYAPP.db.host = "b" │
                    └─────────────────────────────────┘
                              ↓  Watch 订阅
                    ┌─────────────────────────────────┐
                    │  1. 推送当前所有 Key 的最新值      │  ← 初始状态推送
                    │     (db.host="b", db.port="5")   │
                    │  2. 推送 nil（初始值推送完毕标记）  │
                    │  3. 后续实时推送每次变化           │  ← 增量推送
                    └─────────────────────────────────┘

6.2 WatchAll — 监听所有 Key 变化 ​

go
func exampleWatchAll(kv jetstream.KeyValue) {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // WatchAll 监听 Bucket 内所有 Key 的变化
    watcher, err := kv.WatchAll(ctx)
    if err != nil {
        log.Printf("WatchAll 失败: %v", err)
        return
    }
    defer watcher.Stop()

    fmt.Println("开始监听所有 Key 变化...")

    for {
        select {
        case entry, ok := <-watcher.Updates():
            if !ok {
                // Channel 关闭，Watcher 已停止
                fmt.Println("Watcher 已停止")
                return
            }

            if entry == nil {
                // nil 是特殊信号：初始值已全部推送完毕
                // 后续的消息都是实时增量变化
                fmt.Println("=== 初始状态推送完毕，进入实时监听模式 ===")
                continue
            }

            // 处理变化事件
            switch entry.Operation() {
            case jetstream.KeyValuePut:
                fmt.Printf("[PUT]    key=%s  value=%s  rev=%d\n",
                    entry.Key(), string(entry.Value()), entry.Revision())
            case jetstream.KeyValueDelete:
                fmt.Printf("[DELETE] key=%s  rev=%d\n",
                    entry.Key(), entry.Revision())
            case jetstream.KeyValuePurge:
                fmt.Printf("[PURGE]  key=%s  rev=%d\n",
                    entry.Key(), entry.Revision())
            }

        case <-ctx.Done():
            fmt.Println("Context 取消，停止监听")
            return
        }
    }
}

6.3 Watch(pattern) — 通配符监听 ​

go
func exampleWatchPattern(kv jetstream.KeyValue) {
    ctx := context.Background()

    // 只监听 database. 前缀的所有 key
    // NATS 通配符：> 匹配多层，* 匹配单层
    watcher, err := kv.Watch(ctx, "database.>")
    if err != nil {
        log.Printf("Watch 失败: %v", err)
        return
    }
    defer watcher.Stop()

    // 监听特定单层 key（* 匹配单个 token）
    watcherSingle, err := kv.Watch(ctx, "feature.*.enabled")
    if err != nil {
        log.Printf("Watch 失败: %v", err)
        return
    }
    defer watcherSingle.Stop()

    // 监听指定的单个 key
    watcherKey, err := kv.Watch(ctx, "app.version")
    if err != nil {
        log.Printf("Watch 失败: %v", err)
        return
    }
    defer watcherKey.Stop()

    // 处理 database.> 的变化
    go func() {
        for entry := range watcher.Updates() {
            if entry == nil {
                continue // 跳过初始完成信号
            }
            fmt.Printf("数据库配置变化: %s = %s\n",
                entry.Key(), string(entry.Value()))
        }
    }()

    // Watch 选项：IgnoreDeletes 忽略删除事件
    watcherNoDelete, err := kv.WatchAll(ctx, jetstream.IgnoreDeletes())
    if err != nil {
        log.Printf("Watch 失败: %v", err)
        return
    }
    defer watcherNoDelete.Stop()

    // Watch 选项：UpdatesOnly 跳过初始值推送，只接收后续变化
    watcherUpdatesOnly, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
    if err != nil {
        log.Printf("Watch 失败: %v", err)
        return
    }
    defer watcherUpdatesOnly.Stop()

    _ = watcherKey
    _ = watcherSingle

    select {}
}

6.4 KeyValueEntry 结构详解 ​

go
// KeyValueEntry 接口（nats.go jetstream 包）
type KeyValueEntry interface {
    // Bucket 返回所属的 KV Bucket 名称
    Bucket() string

    // Key 返回完整的 key 名称
    Key() string

    // Value 返回当前值的字节数组
    // 如果 Operation 为 Delete/Purge，Value 为 nil
    Value() []byte

    // Revision 返回全局单调递增的版本号（= 底层 Stream Sequence）
    Revision() uint64

    // Delta 在 Watch 场景中表示距最新版本的消息数
    // 初始推送时 Delta > 0，推送完毕后 Delta = 0
    Delta() uint64

    // Created 返回此版本的创建时间
    Created() time.Time

    // Operation 返回此条目的操作类型
    Operation() KeyValueOp // KeyValuePut / KeyValueDelete / KeyValuePurge
}

7. History 版本历史 ​

History 参数控制每个 Key 保留的历史版本数，这是 NATS KV Store 区别于其他 KV 系统的重要特性。

go
func exampleHistory(kv jetstream.KeyValue) {
    ctx := context.Background()

    // 假设 Bucket 的 History=5，以下操作依次写入
    kv.Put(ctx, "app.version", []byte("1.0.0")) // Rev=1
    kv.Put(ctx, "app.version", []byte("1.1.0")) // Rev=2
    kv.Put(ctx, "app.version", []byte("1.2.0")) // Rev=3
    kv.Put(ctx, "app.version", []byte("2.0.0")) // Rev=4
    kv.Put(ctx, "app.version", []byte("2.1.0")) // Rev=5

    // 查询某个 Key 的完整历史
    // History 返回该 Key 的所有历史版本（按时间升序）
    entries, err := kv.History(ctx, "app.version")
    if err != nil {
        log.Printf("History 失败: %v", err)
        return
    }

    fmt.Printf("app.version 的历史版本（共 %d 条）:\n", len(entries))
    for _, entry := range entries {
        fmt.Printf("  Rev=%d  Value=%s  Time=%v  Op=%v\n",
            entry.Revision(),
            string(entry.Value()),
            entry.Created().Format("15:04:05"),
            entry.Operation(),
        )
    }

    // 读取指定版本（回滚或审计用途）
    oldEntry, err := kv.GetRevision(ctx, "app.version", 2)
    if err != nil {
        log.Printf("GetRevision 失败: %v", err)
        return
    }
    fmt.Printf("版本 2 的值: %s\n", string(oldEntry.Value())) // 1.1.0
}

History 使用场景：

场景History 推荐值说明
纯配置存储5~10保留最近几次变更，方便回滚
审计日志50~100追踪配置变更历史
状态机1只需最新状态，节省存储
特征开关3~5快速回滚开关状态

8. 分布式锁实现模式 ​

利用 KV Store 的 Create（CAS）和 Delete 操作，可以实现简单的分布式锁。

go
package lock

import (
    "context"
    "errors"
    "fmt"
    "time"

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

// DistributedLock 基于 NATS KV Store 的分布式锁
type DistributedLock struct {
    kv       jetstream.KeyValue
    lockKey  string
    ownerID  string
    revision uint64
    ttl      time.Duration
}

var ErrLockAlreadyHeld = errors.New("锁已被其他节点持有")
var ErrNotLockOwner = errors.New("当前节点不是锁的持有者")

// TryLock 尝试获取锁（非阻塞）
// ownerID：唯一标识当前节点/协程，例如 "node-1:goroutine-123"
func TryLock(ctx context.Context, kv jetstream.KeyValue, lockKey, ownerID string) (*DistributedLock, error) {
    lockValue := fmt.Sprintf("%s|%d", ownerID, time.Now().UnixNano())

    // Create 是原子 CAS：key 不存在时才写入成功
    rev, err := kv.Create(ctx, lockKey, []byte(lockValue))
    if err != nil {
        if errors.Is(err, jetstream.ErrKeyExists) {
            // 锁已被持有
            entry, getErr := kv.Get(ctx, lockKey)
            if getErr == nil {
                fmt.Printf("锁被持有者 %s 占用\n", string(entry.Value()))
            }
            return nil, ErrLockAlreadyHeld
        }
        return nil, fmt.Errorf("获取锁失败: %w", err)
    }

    return &DistributedLock{
        kv:       kv,
        lockKey:  lockKey,
        ownerID:  ownerID,
        revision: rev,
    }, nil
}

// AcquireLock 阻塞等待获取锁
func AcquireLock(ctx context.Context, kv jetstream.KeyValue, lockKey, ownerID string, retryInterval time.Duration) (*DistributedLock, error) {
    for {
        lock, err := TryLock(ctx, kv, lockKey, ownerID)
        if err == nil {
            return lock, nil
        }
        if !errors.Is(err, ErrLockAlreadyHeld) {
            return nil, err
        }

        // 等待后重试
        select {
        case <-ctx.Done():
            return nil, ctx.Err()
        case <-time.After(retryInterval):
            // 继续重试
        }
    }
}

// Unlock 释放锁
// 只有锁的持有者才能释放（通过检查 Revision 实现）
func (l *DistributedLock) Unlock(ctx context.Context) error {
    // 验证锁仍然是我们持有的（通过 Revision 对比）
    entry, err := l.kv.Get(ctx, l.lockKey)
    if err != nil {
        if errors.Is(err, jetstream.ErrKeyNotFound) {
            return nil // 锁已不存在（可能因 TTL 过期），视为已释放
        }
        return fmt.Errorf("验证锁失败: %w", err)
    }

    if entry.Revision() != l.revision {
        return ErrNotLockOwner // 锁已被其他节点重新获取
    }

    return l.kv.Delete(ctx, l.lockKey)
}

// 使用示例
func UseLock() {
    // kv 已初始化（略）
    var kv jetstream.KeyValue

    ctx := context.Background()

    // 尝试获取锁（带超时）
    lockCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
    defer cancel()

    lock, err := AcquireLock(lockCtx, kv, "lock.critical-section", "node-1", 100*time.Millisecond)
    if err != nil {
        log.Printf("获取锁失败: %v", err)
        return
    }
    defer lock.Unlock(ctx)

    // 执行临界区操作
    fmt.Println("已获取锁，执行临界区操作...")
    time.Sleep(1 * time.Second)
    fmt.Println("临界区操作完成，释放锁")
}

注意事项：

  1. TTL 兜底：如果持锁节点崩溃，锁不会自动释放，除非 Bucket 配置了 TTL。建议把锁 Bucket 的 TTL 设置为锁的最大预期持有时间。
  2. 不是强一致锁：如果网络分区导致客户端与 NATS 集群连接断开，锁可能被其他节点重新获取。对于强一致性要求，建议使用 etcd/Consul。
  3. 续期机制：对于长时间运行的任务，应定期调用 Put 更新锁的值来刷新 TTL（或使用不设 TTL 的方式，依赖显式释放）。

9. 配置中心模式 ​

KV Store + Watch 可以实现配置热更新，无需重启服务即可下发新配置。

go
package config

import (
    "context"
    "encoding/json"
    "log"
    "sync"
    "time"

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

// AppConfig 应用配置结构
type AppConfig struct {
    LogLevel    string        `json:"log_level"`
    Timeout     time.Duration `json:"timeout"`
    MaxConn     int           `json:"max_conn"`
    FeatureFlags map[string]bool `json:"feature_flags"`
}

// ConfigCenter 基于 NATS KV 的配置中心客户端
type ConfigCenter struct {
    kv      jetstream.KeyValue
    mu      sync.RWMutex
    configs map[string]*AppConfig // serviceName -> Config
    callbacks map[string][]func(key string, cfg *AppConfig)
}

func NewConfigCenter(kv jetstream.KeyValue) *ConfigCenter {
    return &ConfigCenter{
        kv:        kv,
        configs:   make(map[string]*AppConfig),
        callbacks: make(map[string][]func(string, *AppConfig)),
    }
}

// LoadAndWatch 加载初始配置并开始监听变化
func (c *ConfigCenter) LoadAndWatch(ctx context.Context, servicePrefix string) error {
    // 监听所有以 servicePrefix 开头的 key
    pattern := servicePrefix + ".>"
    watcher, err := c.kv.Watch(ctx, pattern)
    if err != nil {
        return fmt.Errorf("启动 Watch 失败: %w", err)
    }

    // 在 goroutine 中处理更新
    go func() {
        defer watcher.Stop()
        initialLoadDone := false

        for {
            select {
            case entry, ok := <-watcher.Updates():
                if !ok {
                    log.Println("配置中心 Watcher 已关闭，尝试重连...")
                    // 实际生产中应实现重连逻辑
                    return
                }

                if entry == nil {
                    // 初始值推送完成
                    initialLoadDone = true
                    log.Println("配置中心初始化完成，开始实时监听")
                    continue
                }

                c.handleConfigUpdate(entry, initialLoadDone)

            case <-ctx.Done():
                log.Println("配置中心停止监听")
                return
            }
        }
    }()

    return nil
}

func (c *ConfigCenter) handleConfigUpdate(entry jetstream.KeyValueEntry, isHotUpdate bool) {
    key := entry.Key()

    switch entry.Operation() {
    case jetstream.KeyValuePut:
        var cfg AppConfig
        if err := json.Unmarshal(entry.Value(), &cfg); err != nil {
            log.Printf("配置解析失败 key=%s: %v", key, err)
            return
        }

        c.mu.Lock()
        c.configs[key] = &cfg
        callbacks := c.callbacks[key]
        c.mu.Unlock()

        if isHotUpdate {
            log.Printf("配置热更新: key=%s rev=%d", key, entry.Revision())
        } else {
            log.Printf("配置初始加载: key=%s rev=%d", key, entry.Revision())
        }

        // 触发变更回调
        for _, cb := range callbacks {
            go cb(key, &cfg)
        }

    case jetstream.KeyValueDelete, jetstream.KeyValuePurge:
        c.mu.Lock()
        delete(c.configs, key)
        c.mu.Unlock()
        log.Printf("配置删除: key=%s", key)
    }
}

// Get 获取当前配置（线程安全）
func (c *ConfigCenter) Get(key string) (*AppConfig, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    cfg, ok := c.configs[key]
    return cfg, ok
}

// OnChange 注册配置变更回调
func (c *ConfigCenter) OnChange(key string, callback func(key string, cfg *AppConfig)) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.callbacks[key] = append(c.callbacks[key], callback)
}

// 使用示例
func UseConfigCenter(kv jetstream.KeyValue) {
    ctx := context.Background()
    cc := NewConfigCenter(kv)

    // 注册配置变更回调
    cc.OnChange("myapp.payment-service", func(key string, cfg *AppConfig) {
        log.Printf("支付服务配置已更新: LogLevel=%s MaxConn=%d",
            cfg.LogLevel, cfg.MaxConn)
        // 这里可以触发服务内部的重新初始化
    })

    // 启动监听
    if err := cc.LoadAndWatch(ctx, "myapp"); err != nil {
        log.Fatal(err)
    }

    // 模拟配置下发（运维操作，通常由配置管理工具执行）
    go func() {
        time.Sleep(2 * time.Second)
        cfg := AppConfig{
            LogLevel: "debug",
            Timeout:  30 * time.Second,
            MaxConn:  100,
            FeatureFlags: map[string]bool{
                "new-checkout-flow": true,
            },
        }
        data, _ := json.Marshal(cfg)
        kv.Put(ctx, "myapp.payment-service", data)
    }()

    // 使用配置
    time.Sleep(3 * time.Second)
    if cfg, ok := cc.Get("myapp.payment-service"); ok {
        log.Printf("当前配置: %+v", cfg)
    }
}

10. Object Store 概念 ​

Object Store 是 NATS JetStream 提供的用于存储大型二进制对象的能力，底层同样基于 JetStream Stream，但采用分片（chunked）存储方式：

  • 分片传输：大文件被分割为固定大小的 Chunk 分别存储，避免 NATS 消息大小限制
  • 元数据分离：文件元信息（名称、大小、类型、校验和）存储在独立 Subject
  • 流式 Get：读取时通过 io.Reader 接口流式返回，不需要一次性加载到内存

Object Store 底层结构 ​

创建 Bucket "IMAGES" 后，底层有两个 Stream Subject：

$O.IMAGES.M.>   →  元数据消息（ObjectInfo）
$O.IMAGES.C.>   →  Chunk 数据消息

默认 Chunk 大小为 128KB，可通过 MaxChunkSize 配置调整。

11. Object Store vs KV Store 区别 ​

特性KV StoreObject Store
适用数据大小< 1MB（推荐 < 64KB）任意大小（GB 级别）
存储方式单条消息分片（Chunked）存储
Value 类型[]byteio.Reader / io.Writer
随机读取支持（按 Key）不支持（只能全量读取）
Watch 变化支持支持（元数据级别）
版本历史支持不支持（只有最新版本）
TTL支持支持
CAS 操作支持（Create/Update）不支持
典型用途配置、状态、Feature Flag文件、镜像、模型、证书

12. Object Store 常用操作 ​

go
package main

import (
    "bytes"
    "context"
    "fmt"
    "io"
    "log"
    "os"
    "time"

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

func objectStoreExample() {
    nc, _ := nats.Connect("nats://localhost:4222")
    defer nc.Drain()

    js, _ := jetstream.New(nc)
    ctx := context.Background()

    // 创建 Object Store Bucket
    obs, err := js.CreateOrUpdateObjectStore(ctx, jetstream.ObjectStoreConfig{
        Bucket:      "MODEL_ARTIFACTS",
        Description: "ML 模型文件存储",
        TTL:         30 * 24 * time.Hour, // 30天过期
        Replicas:    3,
        Storage:     jetstream.FileStorage,
        MaxChunkSize: 512 * 1024, // 每个 Chunk 512KB
    })
    if err != nil {
        log.Fatal(err)
    }

    // Put：上传文件（io.Reader 接口）
    file, err := os.Open("/path/to/model.bin")
    if err != nil {
        log.Fatal(err)
    }
    defer file.Close()

    info, err := obs.Put(ctx, &jetstream.ObjectMeta{
        Name:        "model-v2.bin",
        Description: "生产模型 v2.0",
        Headers: nats.Header{
            "Content-Type": []string{"application/octet-stream"},
            "Model-Version": []string{"2.0.0"},
        },
    }, file)
    if err != nil {
        log.Printf("Put 失败: %v", err)
        return
    }
    fmt.Printf("上传成功: name=%s size=%d chunks=%d\n",
        info.Name, info.Size, info.Chunks)

    // Put：直接从 []byte 上传
    data := []byte("小文件内容")
    info2, err := obs.PutBytes(ctx, "config.json", data)
    if err != nil {
        log.Printf("PutBytes 失败: %v", err)
        return
    }
    fmt.Printf("上传成功: name=%s\n", info2.Name)

    // Get：下载文件（返回 io.Reader）
    result, err := obs.Get(ctx, "model-v2.bin")
    if err != nil {
        log.Printf("Get 失败: %v", err)
        return
    }
    defer result.Close()

    // 流式读取
    outFile, _ := os.Create("/tmp/model-v2-download.bin")
    defer outFile.Close()
    n, err := io.Copy(outFile, result)
    if err != nil {
        log.Printf("读取失败: %v", err)
        return
    }
    fmt.Printf("下载完成，共 %d 字节\n", n)

    // 读取对象元数据
    fmt.Printf("对象信息: %+v\n", result.Info())

    // GetBytes：小文件，直接读入内存
    rawData, err := obs.GetBytes(ctx, "config.json")
    if err != nil {
        log.Printf("GetBytes 失败: %v", err)
        return
    }
    fmt.Printf("config.json 内容: %s\n", string(rawData))

    // GetString：读取为字符串
    str, err := obs.GetString(ctx, "config.json")
    if err != nil {
        log.Printf("GetString 失败: %v", err)
        return
    }
    fmt.Printf("config.json 字符串: %s\n", str)

    // List：列出所有对象的元数据
    objects, err := obs.List(ctx)
    if err != nil {
        log.Printf("List 失败: %v", err)
        return
    }
    for _, obj := range objects {
        fmt.Printf("  %s: size=%d, modified=%v\n",
            obj.Name, obj.Size, obj.ModTime)
    }

    // Delete：删除对象
    err = obs.Delete(ctx, "model-v2.bin")
    if err != nil {
        log.Printf("Delete 失败: %v", err)
        return
    }
    fmt.Println("对象已删除")

    // Watch：监听对象变化
    watcher, err := obs.Watch(ctx)
    if err != nil {
        log.Printf("Watch 失败: %v", err)
        return
    }
    defer watcher.Stop()

    go func() {
        for info := range watcher.Updates() {
            if info == nil {
                fmt.Println("Object Store Watch 初始化完成")
                continue
            }
            if info.Deleted {
                fmt.Printf("对象删除: %s\n", info.Name)
            } else {
                fmt.Printf("对象更新: %s size=%d\n", info.Name, info.Size)
            }
        }
    }()

    // PutFile：上传本地文件（便捷方法）
    _, err = obs.PutFile(ctx, "/path/to/certificate.pem")
    if err != nil {
        // 文件不存在则忽略此错误
        _ = err
    }

    // GetFile：下载到本地文件
    err = obs.GetFile(ctx, "certificate.pem", "/tmp/cert.pem")
    if err != nil {
        _ = err
    }

    _ = bytes.NewBuffer(nil) // 避免未使用的 import
}

13. 完整代码示例 ​

13.1 基础 CRUD ​

go
package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "time"

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

func main() {
    nc, err := nats.Connect("nats://localhost:4222",
        nats.Name("kv-example"),
    )
    if err != nil {
        log.Fatal("连接 NATS 失败:", err)
    }
    defer nc.Drain()

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

    ctx := context.Background()

    // 创建 KV Bucket
    kv, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
        Bucket:       "DEMO",
        History:      5,
        TTL:          1 * time.Hour,
        MaxValueSize: 64 * 1024, // 64KB
    })
    if err != nil {
        log.Fatal("创建 KV Bucket 失败:", err)
    }
    fmt.Println("KV Bucket 创建成功")

    // Put
    rev1, _ := kv.Put(ctx, "greeting", []byte("Hello"))
    fmt.Printf("Put greeting, Rev=%d\n", rev1)

    rev2, _ := kv.Put(ctx, "greeting", []byte("Hello, NATS!"))
    fmt.Printf("Put greeting, Rev=%d\n", rev2)

    // Get
    entry, err := kv.Get(ctx, "greeting")
    if err != nil {
        log.Fatal("Get 失败:", err)
    }
    fmt.Printf("Get greeting: value=%s rev=%d\n", string(entry.Value()), entry.Revision())

    // Create（幂等写入）
    _, err = kv.Create(ctx, "unique-key", []byte("only-once"))
    if err != nil {
        if errors.Is(err, jetstream.ErrKeyExists) {
            fmt.Println("Create 失败：key 已存在")
        }
    } else {
        fmt.Println("Create 成功")
    }

    // Update（CAS）
    newRev, err := kv.Update(ctx, "greeting", []byte("Updated!"), entry.Revision())
    if err != nil {
        fmt.Printf("Update 失败: %v\n", err)
    } else {
        fmt.Printf("Update 成功, newRev=%d\n", newRev)
    }

    // Keys
    keys, _ := kv.Keys(ctx)
    fmt.Printf("所有 Key: %v\n", keys)

    // History
    history, _ := kv.History(ctx, "greeting")
    fmt.Printf("greeting 历史版本数: %d\n", len(history))
    for _, h := range history {
        fmt.Printf("  Rev=%d Value=%s Op=%v\n", h.Revision(), string(h.Value()), h.Operation())
    }

    // Delete
    kv.Delete(ctx, "greeting")
    _, err = kv.Get(ctx, "greeting")
    fmt.Printf("删除后 Get 错误: %v\n", err) // ErrKeyNotFound

    // Purge
    kv.Put(ctx, "to-purge", []byte("data"))
    kv.Purge(ctx, "to-purge")
    fmt.Println("Purge 完成")
}

13.2 Watch 监听示例 ​

go
package main

import (
    "context"
    "fmt"
    "log"
    "time"

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

func watchDemo() {
    nc, _ := nats.Connect("nats://localhost:4222")
    defer nc.Drain()

    js, _ := jetstream.New(nc)
    ctx := context.Background()

    kv, _ := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
        Bucket:  "WATCH_DEMO",
        History: 3,
    })

    // 预先写入一些数据
    kv.Put(ctx, "config.a", []byte("value-a"))
    kv.Put(ctx, "config.b", []byte("value-b"))
    kv.Put(ctx, "other.key", []byte("other"))

    // 监听 config.> 前缀
    watcher, err := kv.Watch(ctx, "config.>")
    if err != nil {
        log.Fatal(err)
    }

    // 在另一个 goroutine 中模拟配置变更
    go func() {
        time.Sleep(2 * time.Second)
        kv.Put(ctx, "config.a", []byte("updated-a"))
        time.Sleep(1 * time.Second)
        kv.Delete(ctx, "config.b")
        time.Sleep(1 * time.Second)
        kv.Put(ctx, "config.c", []byte("new-c"))
    }()

    // 处理 Watch 事件
    timeout := time.After(8 * time.Second)
    initialDone := false

    for {
        select {
        case entry, ok := <-watcher.Updates():
            if !ok {
                fmt.Println("Watcher 关闭")
                return
            }
            if entry == nil {
                initialDone = true
                fmt.Println(">>> 初始状态推送完毕，进入实时监听 <<<")
                continue
            }

            prefix := "[初始]"
            if initialDone {
                prefix = "[热更新]"
            }

            switch entry.Operation() {
            case jetstream.KeyValuePut:
                fmt.Printf("%s PUT  key=%-20s value=%s\n",
                    prefix, entry.Key(), string(entry.Value()))
            case jetstream.KeyValueDelete:
                fmt.Printf("%s DEL  key=%s\n", prefix, entry.Key())
            case jetstream.KeyValuePurge:
                fmt.Printf("%s PURGE key=%s\n", prefix, entry.Key())
            }

        case <-timeout:
            fmt.Println("Demo 结束")
            watcher.Stop()
            return
        }
    }
}

13.3 分布式锁完整示例 ​

go
package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "sync"
    "time"

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

func distributedLockDemo() {
    nc, _ := nats.Connect("nats://localhost:4222")
    defer nc.Drain()

    js, _ := jetstream.New(nc)
    ctx := context.Background()

    // 锁 Bucket：TTL=30s 作为兜底（防止持锁节点崩溃）
    kv, _ := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
        Bucket: "LOCKS",
        TTL:    30 * time.Second,
    })

    var wg sync.WaitGroup
    results := make(chan string, 10)

    // 模拟 5 个节点并发抢锁
    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go func(nodeID int) {
            defer wg.Done()

            ownerID := fmt.Sprintf("node-%d", nodeID)
            lockKey := "lock.critical-job"

            // 尝试获取锁
            rev, err := kv.Create(ctx, lockKey, []byte(ownerID))
            if err != nil {
                if errors.Is(err, jetstream.ErrKeyExists) {
                    results <- fmt.Sprintf("[node-%d] 未能获取锁，放弃", nodeID)
                    return
                }
                results <- fmt.Sprintf("[node-%d] 错误: %v", nodeID, err)
                return
            }

            // 成功获取锁
            results <- fmt.Sprintf("[node-%d] 成功获取锁 rev=%d，开始执行任务", nodeID, rev)

            // 执行临界区工作
            time.Sleep(500 * time.Millisecond)

            // 释放锁（通过检查 Revision 确保是自己持有的锁）
            entry, err := kv.Get(ctx, lockKey)
            if err == nil && entry.Revision() == rev {
                kv.Delete(ctx, lockKey)
                results <- fmt.Sprintf("[node-%d] 任务完成，释放锁", nodeID)
            }
        }(i)
    }

    // 等待所有 goroutine 完成
    go func() {
        wg.Wait()
        close(results)
    }()

    // 打印结果
    for msg := range results {
        fmt.Println(msg)
    }
}

func main() {
    distributedLockDemo()
}

13.4 配置热更新完整示例 ​

go
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "sync/atomic"
    "time"
    "unsafe"

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

type ServiceConfig struct {
    LogLevel    string `json:"log_level"`
    MaxWorkers  int    `json:"max_workers"`
    DebugMode   bool   `json:"debug_mode"`
    UpdatedAt   string `json:"updated_at"`
}

// AtomicConfig 原子配置指针，实现无锁读取
type AtomicConfig struct {
    p unsafe.Pointer
}

func (ac *AtomicConfig) Load() *ServiceConfig {
    return (*ServiceConfig)(atomic.LoadPointer(&ac.p))
}

func (ac *AtomicConfig) Store(cfg *ServiceConfig) {
    atomic.StorePointer(&ac.p, unsafe.Pointer(cfg))
}

func hotConfigDemo() {
    nc, _ := nats.Connect("nats://localhost:4222")
    defer nc.Drain()

    js, _ := jetstream.New(nc)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // 初始化配置 Bucket
    kv, _ := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
        Bucket:  "SERVICE_CONFIG",
        History: 5,
    })

    // 写入初始配置
    initialCfg := ServiceConfig{
        LogLevel:   "info",
        MaxWorkers: 10,
        DebugMode:  false,
        UpdatedAt:  time.Now().Format(time.RFC3339),
    }
    data, _ := json.Marshal(initialCfg)
    kv.Put(ctx, "myservice", data)

    // 原子配置对象
    var atomicCfg AtomicConfig
    atomicCfg.Store(&initialCfg)

    // 启动 Watch
    watcher, err := kv.Watch(ctx, "myservice")
    if err != nil {
        log.Fatal(err)
    }

    go func() {
        defer watcher.Stop()
        for entry := range watcher.Updates() {
            if entry == nil || entry.Operation() != jetstream.KeyValuePut {
                continue
            }
            var cfg ServiceConfig
            if err := json.Unmarshal(entry.Value(), &cfg); err != nil {
                log.Printf("配置解析失败: %v", err)
                continue
            }
            atomicCfg.Store(&cfg)
            log.Printf("配置已热更新: LogLevel=%s MaxWorkers=%d DebugMode=%v",
                cfg.LogLevel, cfg.MaxWorkers, cfg.DebugMode)
        }
    }()

    // 模拟配置下发
    go func() {
        time.Sleep(2 * time.Second)
        newCfg := ServiceConfig{
            LogLevel:   "debug",
            MaxWorkers: 20,
            DebugMode:  true,
            UpdatedAt:  time.Now().Format(time.RFC3339),
        }
        data, _ := json.Marshal(newCfg)
        kv.Put(ctx, "myservice", data)

        time.Sleep(2 * time.Second)
        revertCfg := ServiceConfig{
            LogLevel:   "info",
            MaxWorkers: 10,
            DebugMode:  false,
            UpdatedAt:  time.Now().Format(time.RFC3339),
        }
        data, _ = json.Marshal(revertCfg)
        kv.Put(ctx, "myservice", data)
    }()

    // 模拟业务循环，读取配置（无锁，极低开销）
    for i := 0; i < 6; i++ {
        time.Sleep(1 * time.Second)
        cfg := atomicCfg.Load()
        fmt.Printf("[t+%ds] LogLevel=%s MaxWorkers=%d DebugMode=%v\n",
            i+1, cfg.LogLevel, cfg.MaxWorkers, cfg.DebugMode)
    }
}

func main() {
    hotConfigDemo()
}

总结 ​

功能API特点
写入/更新Put无条件写入，返回新 Revision
读取Get / GetRevision支持读取历史版本
条件写入CreateKey 不存在才成功（用于分布式锁）
乐观锁更新Update(rev)版本号匹配才成功（用于并发安全更新）
软删除Delete追加删除标记，保留历史
清除历史Purge释放存储空间
实时推送Watch / WatchAll初始值 + 增量变化推送
历史查询History返回指定 Key 的所有历史版本
列出 KeyKeys只返回存活的 Key