在我们开发路边智能摄像头、门店自助终端这类边缘设备时，经常会碰到网络不给力的情况——比如雨天基站信号弱、门店临时断网检修，这时候如果依赖NATS集群的实时通信，要么摄像头的报警指令收不到，要么设备采集的温度数据传不上去，业务就会失效。这时候就需要给Edge端的NATS LeafNode节点做一套离线消息缓存策略，断网时本地存消息，恢复网络后自动同步，保障数据不丢。
一、为什么Edge端NATS LeafNode要做离线缓存？
1.1 边缘设备的网络特性
和机房里的服务器不同，Edge设备大多部署在户外或人流大的商业场景，网络状态波动极大：可能刚连接上NATS集群，几秒钟后就断网，几分钟后又恢复。如果不做缓存，设备的指令下发、数据上报就会“失联”，比如连锁门店的智能冷柜，后台要调整温度阈值，断网时指令收不到，冷柜就会一直用旧的阈值，不仅费电还影响食材保存。
1.2 NATS LeafNode的核心角色
NATS LeafNode相当于边缘设备接入NATS主集群的“入口”，设备不用自己建集群，直接作为LeafNode连主集群，就能实现跨设备的消息通信。但Edge端的LeafNode没法像机房服务器那样长期稳定联网，所以必须本地缓存消息，等网络恢复后再同步到主集群。
二、Edge端离线消息缓存的实现（Node.js技术栈）
这里我们用轻量的技术组合：Node.js 18.x + NATS官方JS客户端 + lowdb（适合小数据的本地缓存库），代码会全程带注释，方便开发者上手。
// 技术栈：Node.js 18.x + nats@2.16.0（NATS官方JS客户端） + lowdb@3.0.1（轻量本地JSON缓存）
import { connect, StringCodec } from 'nats'; // NATS核心依赖，用于连接和编解码消息
import { Low, JSONFile } from 'lowdb'; // lowdb依赖，用于本地持久化缓存
// 1. 初始化本地缓存数据库，所有离线消息存在这里
const cacheAdapter = new JSONFile('edge_offline_cache.json'); // 缓存文件，存在设备本地
const cacheDB = new Low(cacheAdapter);
await cacheDB.read(); // 读取本地缓存
// 如果缓存不存在，初始化空的待同步消息数组
cacheDB.data = cacheDB.data || { pending_msgs: [] };
await cacheDB.write(); // 初始化完成，保存到本地
// 2. 全局变量，用于存储NATS连接实例
let natsConnection = null;
const stringCodec = StringCodec(); // 消息编解码工具，把二进制转成可读文本
// 3. 连接NATS LeafNode的函数，断网时会定期重试
async function connectNATSLeaf() {
try {
// 连接NATS主集群，这里替换成你的集群地址
natsConnection = await connect({ servers: 'nats://your-nats-main-cluster:4222' });
console.log('NATS LeafNode连接成功，开始检查待同步消息');
// 连接成功后，先把本地缓存的所有消息同步到主集群
await syncPendingMessages();
// 订阅需要的主题，比如设备数据上报主题，主题名根据业务改
const dataSub = natsConnection.subscribe('edge.device.sensor.data');
// 循环处理订阅到的消息
(async () => {
for await (const msg of dataSub) {
// 收到消息后，根据联网状态处理：联网就直接发主集群，否则缓存
if (natsConnection?.isConnected) {
await processReceivedMsg(msg);
} else {
await cacheOfflineMsg(msg);
}
}
})();
} catch (err) {
console.error('NATS LeafNode连接失败，进入离线缓存模式：', err.message);
natsConnection = null;
}
}
// 4. 处理正常联网时收到的消息（比如上报数据给主集群）
async function processReceivedMsg(msg) {
const msgContent = stringCodec.decode(msg.data);
console.log('联网状态，处理消息：', msgContent);
// 把消息上报到NATS主集群的业务主题
await natsConnection.publish('main.edge.data.upload', msg.data);
}
// 5. 缓存断网时收到的消息，存在本地
async function cacheOfflineMsg(msg) {
const msgObj = {
id: Date.now() + Math.random().toString(36).slice(2), // 生成唯一消息ID，用于去重和删除
subject: msg.subject, // 消息的主题，方便同步时转发
data: stringCodec.decode(msg.data), // 消息内容
createTime: Date.now() // 消息创建时间，保证同步顺序
};
// 把消息加到缓存数组
cacheDB.data.pending_msgs.push(msgObj);
await cacheDB.write(); // 持久化到本地文件
console.log('断网，消息已缓存：', msgObj.id);
}
// 6. 同步本地缓存的所有消息到主集群，连网后自动调用
async function syncPendingMessages() {
if (!natsConnection || !natsConnection.isConnected || cacheDB.data.pending_msgs.length === 0) return;
console.log(`开始同步${cacheDB.data.pending_msgs.length}条离线消息`);
// 按时间顺序同步（先存的先发，保证业务时序）
for (const msg of cacheDB.data.pending_msgs) {
try {
// 把缓存的消息转发到主集群对应主题
await natsConnection.publish(`main.${msg.subject}`, stringCodec.encode(msg.data));
console.log(`同步消息成功：${msg.id}`);
// 同步成功后，从本地缓存删除该消息
const msgIndex = cacheDB.data.pending_msgs.findIndex(m => m.id === msg.id);
if (msgIndex !== -1) cacheDB.data.pending_msgs.splice(msgIndex, 1);
} catch (err) {
console.error(`同步消息失败：${msg.id}，将在下次重试`, err.message);
// 同步失败的消息会留在缓存，下次重连再同步
}
}
// 同步完成后，保存更新后的缓存
await cacheDB.write();
}
// 主逻辑：每5秒检查一次NATS连接，自动重连+同步缓存
setInterval(async () => {
if (!natsConnection || !natsConnection.isConnected) {
console.log('尝试重连NATS LeafNode...');
await connectNATSLeaf();
} else {
// 联网状态下，同步未完成的缓存消息
await syncPendingMessages();
}
}, 5000);
2.1 缓存策略的核心规则
这套缓存方案用了两个核心规则，适配Edge设备的特点：一是消息时序优先，同步时按消息创建时间先后发送，避免主集群收到的消息乱序；二是缓存上限控制，可以在代码里加判断，比如if (cacheDB.data.pending_msgs.length > 500) cacheDB.data.pending_msgs.shift()，删最早的消息，防止本地存储被占满（Edge设备一般只有几百MB的存储，不能无限存）。
2.2 断网恢复的同步逻辑
代码里用了定时轮询的方式，每5秒检查一次NATS连接状态：没联网就尝试重连，连成功就自动同步所有未发的缓存消息，不需要人工干预，符合边缘设备“无人值守”的需求。
三、离线缓存策略的技术优缺点
3.1 优点
- 数据零丢失：断网期间的所有消息都存在本地，恢复后自动同步，不会因为网络波动丢数据；
- 轻量适配Edge：用的lowdb只有几十KB，Node.js运行时的内存占用极低，适合性能弱的边缘设备（比如低功耗的ARM设备）；
- 兼容NATS标准：完全基于NATS LeafNode的官方规范，不用改主集群的配置，接入成本低；
3.2 缺点
- 依赖本地存储可靠性：如果Edge设备的本地存储（比如SD卡）损坏，缓存就会丢失，所以实际项目中可以把重要消息同时备份到云端的临时存储；
- 同步延迟风险：如果网络故障时间太长，缓存的消息太多，同步的时候会占用带宽，可能影响其他业务，所以要设缓存上限，或者给消息加优先级（比如报警消息优先级高，温度上报优先级低，优先同步高优先级消息）。
四、实际应用场景
4.1 连锁便利店智能冷柜
每个冷柜是Edge端的NATS LeafNode，需要定期上报温度数据给后台，同时接收后台下发的温度调整指令。如果门店网络断了，冷柜会把采集到的温度缓存，同时接收后台的调整指令也缓存，网好后自动同步，保证冷柜的温度监控和调整不中断。
4.2 小区智能充电桩
充电桩需要上报充电状态、计费数据给运营商，同时接收远程启动/停止充电的指令。断网时，充电数据和指令都缓存，恢复后同步，避免计费错误和远程控制失效，保障用户和运营商的权益。
五、注意事项
5.1 缓存大小的上限控制
Edge设备的存储有限，必须设置缓存的最大条数，比如最多存500条消息，超过就删最早的，或者按优先级删，不要让缓存占满所有存储空间，导致设备其他功能用不了。
5.2 同步的可靠性保障
对于关键业务（比如充电桩的计费数据），可以把NATS的publish改成request模式，发消息后等主集群回复确认，再删除本地缓存，避免消息没发成功就删了，但要注意request会增加延迟，适合对可靠性要求高的场景，一般数据用publish就够了。
5.3 资源占用的平衡
不要把定时检查的间隔设得太短（比如小于1秒），会频繁占用CPU；同步消息的时候可以批量处理，比如一次同步10条，减少网络请求的开销，避免给主集群造成压力。
六、总结
Edge端NATS LeafNode的离线消息缓存策略，是边缘计算场景里保障通信稳定性的关键方案，核心是“本地存、自动连、恢复后同步”，用轻量的技术实现适配Edge设备的弱网络环境，同时要注意缓存大小、同步可靠性和资源占用这三个关键点，才能让方案在实际项目中稳定运行。
评论
围绕“Edge端NATS LeafNode的离线消息缓存策略与断网恢复后同步”参与讨论