别再手写“断点续传”了！用 NATS JetStream Leaf Node 构建“永不丢数”的边缘同步网路

架构师笔记

2026-02-05 13:30:00

#NATS #JetStream #边缘同步 #LeafNode #IIoT #数据一致性 #替代K

一、场景痛点：MQTT 桥接的“性能天花板”

在最近的一个车载数据上云项目（500 辆矿卡，每辆车每秒产生 2000 条 CAN 总线日志）中，我们遇到了 MQTT 的瓶颈：

现状：车端使用 MQTT Broker (Mosquitto) 进行桥接。
问题：

弱网堆积：矿区 5G 信号差，车辆经常断网 30 分钟。Mosquitto 的桥接队列在内存中堆积，不仅吃光内存，而且在网络恢复瞬间，数百万条积压消息的并发推送导致云端 Broker CPU 飙升 100%，引发雪崩效应。
顺序错乱：MQTT QoS 1 并不严格保证全局顺序，导致云端收到的车辆轨迹点是乱序的（先收到 10:02 的点，后收到 10:01 的点）。
运维复杂：为了防止丢数据，开发团队自己在应用层写了一套复杂的 SQLite -> Check Network -> Upload 逻辑，Bug 频出。

架构师指令：放弃在应用层造轮子。

我们需要一个“天然支持存储转发、流量削峰、且对应用透明”的协议层解决方案。NATS JetStream 的 Leaf Node 拓扑就是为此而生的。

二、架构设计：Leaf Node (叶子节点) 拓扑

NATS 的 Leaf Node 是一种特殊的路由模式，它允许边缘端的 NATS Server 作为“叶子”，连接到云端的“枢纽（Hub）”。

本地闭环：当断网时，边缘应用连接本地 Leaf Node，消息写入本地磁盘（JetStream 持久化），应用端感知不到断网，发送速度不受影响。
透明同步：Leaf Node 会自动维护与 Cloud Hub 的连接。一旦联网，它会自动将本地 Stream 中的新数据同步到云端，自带流控（Flow Control），不会打垮云端。
安全性：Leaf Node 发起出站连接（Outbound），边缘侧无需开放任何入站端口，防火墙友好。

拓扑图：

车载应用 -> [本地 NATS Leaf (JetStream 磁盘)] --(自动同步/流量整形)--> [云端 NATS Hub] -> 数据湖

三、核心实施步骤 (Copy & Paste)

1. 云端 Hub 配置 (hub.conf)

首先配置云端 NATS 服务器，启用 JetStream 并接受叶子节点连接。

Text

# 开启 JetStream 持久化
jetstream {
    store_dir: "/data/nats/hub"
    max_mem: 1G
    max_file: 100G
}

# 监听 Leaf Node 连接
leafnodes {
    port: 7422
    no_auth_user: "leaf_user" # 简化示例，生产环境请用 NKEYS
}

listen: 4222

2. 边缘端 Leaf 配置 (edge.conf)

这是核心。配置它主动连接云端，并定义这就即是一个“本地集群”也是“云端的延伸”。

Text

server_name: "truck-001"

jetstream {
    store_dir: "/var/lib/nats/edge"
    max_file: 5G  # 本地磁盘缓存上限，超限自动丢弃最旧数据
}

leafnodes {
    remotes = [
        {
            url: "nats://hub.example.com:7422"
            # 将云端的 "cloud.inbox" 映射到本地，反之亦然
        }
    ]
}

3. 创建 Stream (流定义)

使用 NATS CLI 工具，在云端定义一个流，它会自动通过 Leaf Node 延伸到边缘。

Bash

# 在边缘端执行，创建一个镜像流
nats stream add VEHICLE_DATA \
  --subjects "vehicle.*" \
  --storage file \
  --retention limits \
  --max-age 7d \
  --replicas 1

4. 业务代码 (Go 示例)

应用层代码极其简单，完全不需要写重试逻辑。

package main

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

func main() {
    // 1. 连接本地 Leaf Node (断网也能连上！)
	nc, _ := nats.Connect("nats://localhost:4222")
	js, _ := nc.JetStream()

    // 2. 发送数据
    // 即使拔掉网线，这里也会瞬间返回 ACK (因为写入了本地磁盘)
    // NATS 后台负责在联网时把数据传给云端
	_, err := js.Publish("vehicle.sensor", []byte("data payload"))
	if err != nil {
		log.Fatal(err)
	}
}

四、踩坑复盘 (Red Flags)

1. 磁盘 IOPS 瓶颈

现象：车辆产生大量高频振动数据时，NATS 报错 Slow Consumer 或写入延迟高。
原因：JetStream 是写磁盘的。如果网关用的是普通的 TF 卡，IOPS 只有几百，扛不住每秒几千条的持久化写入。
对策：

硬件：必须使用 NVMe SSD 或 工业级 eMMC。
配置：对于非关键数据（如 debug 日志），使用 Mem 模式（内存流），不落盘。

2. 消费者“回溯”风暴

现象：云端消费者重启后，突然收到了几天前的旧数据。
原因：JetStream 默认是持久化的，消费者如果没有正确设置 DeliverPolicy（投递策略），会从流的起点开始重放。
对策：消费者订阅时明确指定 DeliverNew (只收新数据) 或 DeliverLast (只收最后一条)。

3. Subject 设计层级过深

现象：使用了 factory.line1.machine2.sensor3.temp.value... 这样 10 层的主题，导致匹配性能下降。
建议：NATS 推荐主题层级控制在 16 层以内，且尽量把高基数（High Cardinality）的 ID 放在主题的后面部分，如 metrics.us-east.sensor-001。

五、关联资源与选型

NATS 是 Go 写的，单二进制文件，资源占用极低，非常适合边缘。