第 17 篇：离线缓存设计——多文件循环队列保障断网不丢数据

前端流控挡住了能丢的和能慢的但当网络彻底断开时所有发往云端的消息都必须落地存下来。为什么不用 SQLite为什么不用内存队列本文设计一套基于自定义文件格式的循环队列——追加写入、指针移动即删除、TPS 限速消费、带容量预警。一、开篇场景剩下来的不能丢——断网 3 天存不存得住第 14 篇分了三级LOW 可丢第 15 篇设了令牌桶过载了守住入口第 16 篇做了慢 ACK 背压不能丢的让上游减速。这三道防线连起来处理了一条 MEDIUM/HIGH 消息从设备上报到路由分发的完整路径。但还有一个场景没有覆盖——网络彻底断了。背压能减速减速不了断网。当光缆被挖断所有发往云端的 HIGH 级别消息MQTT PUBACK 正常回但云端 Client 发不出去。这些消息不能丢必须落地。工厂因为施工挖断了主干光缆。运营商的维修承诺是 72 小时。这 72 小时里边缘网关上的 500 个传感器还在不停地产生数据。你的离线缓存需要扛住什么写入量500 传感器 × 1 条/秒 × 200 字节 100KB/s ≈8.6GB/天缓存总容量SD 卡分配给离线缓存的空间 32GB3 天数据量约 26GB——能存下但很紧消费速度网络恢复后要以可控的速度补传不能瞬间把云打爆挑战数据持续写入容量有限可能溢出消费要限速消息有先后顺序FIFO断电不能丢数据。本文涉及的 Go 包timesyncsync/atomiccontextgolang.org/x/time/rate二、概念铺垫为什么不直接用 SQLite很多人的第一反应是——“用 SQLite 存消息简单。”但 SQLite 在这个场景下有两个致命问题问题SQLite 表现自定义文件队列 表现写入开销每行 INSERT 都有事务日志、B-tree 更新、页分配追加写入seek write一次系统调用删除开销DELETE 产生碎片需要 VACUUM 回收空间移动文件头指针零碎片并发写入单写锁线程安全靠互斥单 goroutine 写入 channel天然无锁突发性能1000 TPS 写入时 B-tree 频繁分裂追加写入接近磁盘顺序写极限简单说SQLite 是通用数据库我们的离线缓存是专用消息队列。用一个设计为通用的工具做专用的事情性能差一个数量级是很正常的。三、方案设计多文件循环队列3.1 文件队列结构db/offline_cache/ ├── queue_v5_1704067200000 ← 最早的文件队首消费端 ├── queue_v5_1704153600000 ← 中间文件 ├── queue_v5_1704240000000 ← 中间文件 └── queue_v5_1704330000000 ← 最新文件队尾写入端每个文件固定 32MB。文件内是一个环形缓冲区┌───────── Header (64字节) ───────┬── Data Area (剩余) ────────┐ │ first_pos: 指向第一条消息偏移 │ msg1 │ msg2 │ msg3 │ free │ │ last_pos: 指向最后一条消息偏移 │ │ │ │ │ │ count: 当前文件内消息数量 │◀──── 有效数据 ──▶│◀ 空闲 ▶│ │ file_size: 文件总大小(32MB) │ │ └─────────────────────────────────┴──────────────────────────┘3.2 异步写入消息到达 → 不是直接写文件而是先写入一个缓冲 channel容量 1000由专门的写入 goroutine 消费typeOfflineCachestruct{writeChchan*CacheMessage// 缓冲 channelqueues[]*QueueFile// 多文件队列按时间排序opt CacheOption limitStatus atomic.Int32// 容量状态CAS 保证并发安全notifyFnfunc(levelstring)// 容量通知回调}func(c*OfflineCache)Write(msg*CacheMessage){select{casec.writeCh-msg:// 写入成功default:// channel 满——说明写入速度远超磁盘 IO 速度// 按消息的可靠性级别决策// LOW → 直接丢弃第 14 篇的策略// MEDIUM/HIGH → 阻塞等待不能丢ifmsg.LevelReliabilityLow{return}c.writeCh-msg// 阻塞等待}}3.3 写入循环func(c*OfflineCache)writeLoop(){formsg:rangec.writeCh{// 拿到队尾文件最新的文件tailQ:c.queues[len(c.queues)-1]// 序列化消息data:serialize(msg)// 如果当前文件剩余空间不够 → 创建新文件iftailQ.FreeSpace()len(data)headerSize{newFile:c.createNewQueueFile()c.queuesappend(c.queues,newFile)tailQnewFile}// 追加写入tailQ.Append(data)// 检查容量是否触发阈值c.checkCapacity()}}3.4 FIFO 淘汰——容量不够时删最旧的func(c*OfflineCache)ensureCapacity(neededBytesint64){for{usedBytes:c.totalUsedBytes()ifusedBytesneededBytesc.opt.MaxCapacity{return// 容量够}// 从队首文件淘汰最旧的消息frontQ:c.queues[0]// 一次删 10 条批量操作减少开销frontQ.Remove(10)// 如果队首文件被清空 → 删除它iffrontQ.Count()0{frontQ.Delete()c.queuesc.queues[1:]}}}3.5 消费者——TPS 限速补传网络恢复后离线的消息需要补传到云端。但不能瞬间全发过去——会打爆云端 API。func(c*OfflineCache)consumeLoop(){limiter:rate.NewLimiter(rate.Limit(c.opt.ConsumeTPS),c.opt.ConsumeTPS)for{// 从队首文件读取一批消息frontQ:c.queues[0]batch,_:frontQ.Peek(c.opt.BatchSize)iflen(batch)0{// 当前文件没有消息了iflen(c.queues)1{// 这是最后一个文件且没有新消息 → 休眠等待time.Sleep(5*time.Second)continue}// 不是最后一个文件 → 删除空文件继续消费下一个frontQ.Delete()c.queuesc.queues[1:]continue}// 过滤过期消息超过 7 天的消息直接丢弃batchfilterExpired(batch,c.opt.MaxAge)// 限速消费每个 goroutine 发一条varwg sync.WaitGroupfor_,msg:rangebatch{limiter.Wait(context.Background())// 等令牌wg.Add(1)gofunc(m*CacheMessage){deferwg.Done()c.consumeFunc(m)// 根据可靠性级别调用不同的重试策略}(msg)}wg.Wait()// 消费成功 → 从文件移除frontQ.Remove(len(batch))}}3.6 容量上下限通知——防止缓存溢出┌──────────────────────────────────────┐ │ 用量 60% │ 状态: UnLimit正常 │ │ ├──────────────────────────────────────┤ │ 70% 上限阈值 │ ← 超过触发 upper_limit 通知 │ │ 状态变为 Limit │ 60% 下限阈值 │ ← 降到触发 lower_limit 通知 │ │ 状态恢复 UnLimit ├──────────────────────────────────────┤ │ 用量 100% │ 此时 FIFO 淘汰机制已在不断删旧数据 └──────────────────────────────────────┘为什么用 70%/60% 两组阈值而不是 80%/80%这是滞回Hysteresis设计。如果上下限都是 70%当用量在 70% 附近波动时会不断发出满了→恢复→满了→恢复的通知。70% 和 60% 之间有 10% 的缓冲区间只有用量真正降下来了才通知恢复。func(c*OfflineCache)checkCapacity(){used:c.totalUsedBytes()upperLimit:int64(float64(c.opt.MaxCapacity)*0.7)lowerLimit:int64(float64(c.opt.MaxCapacity)*0.6)// 触发上限——从正常切到限流ifusedupperLimitc.limitStatus.CompareAndSwap(UnLimit,Limit){c.notifyFn(upper_limit)// 通知模块停止发送}// 触发下限——从限流切到正常ifusedlowerLimitc.limitStatus.CompareAndSwap(Limit,UnLimit){c.notifyFn(lower_limit)// 通知模块恢复发送}}3.7 消息老化——自动清理过期数据网络断了 30 天后恢复——这 30 天之前的温度数据已经毫无价值了你不可能分析一个月前的温度异常。通过MaxAge配置自动清理过期消息func(c*OfflineCache)cleanExpiredLoop(){ticker:time.NewTicker(24*time.Hour)forrangeticker.C{for_,q:rangec.queues{// 遍历文件中的所有消息删除超过 MaxAge 的q.RemoveExpired(c.opt.MaxAge)}}}四、Go 核心骨架funcNewOfflineCache(opt CacheOption)*OfflineCache{cache:OfflineCache{writeCh:make(chan*CacheMessage,1000),opt:opt,}cache.loadExistingFiles()// 加载磁盘上已有的队列文件gocache.writeLoop()gocache.consumeLoop()gocache.cleanExpiredLoop()returncache}typeCacheOptionstruct{MaxCapacityint64// 总容量上限字节如 32GBMaxAge time.Duration// 消息老化时间默认 7 天ConsumeTPSint// 消费速率条/秒默认 100BatchSizeint// 每次读取的批次大小默认 50}// 一条消息在缓存中的完整生命周期// 1. Write(msg) → writeCh// 2. writeLoop → 追加到队尾文件// 3. 如果容量不够 → FIFO 淘汰队首旧消息// 4. consumeLoop → 从队首文件批量读取// 5. 限速消费 → 发往云端// 6. 消费成功 → 文件指针前移// 7. 文件清空 → 删除文件五、边界与反模式反模式一内存队列代替文件队列错误做法用内存 channel 或 slice 做离线缓存。为什么错内存中的数据断电就丢。而且 32GB 的数据放内存里——你的边缘网关总共才 4GB 内存。反模式二一个文件无限增长错误做法所有消息追加到一个文件文件无限变大。为什么错文件越大seek 越慢。32MB 一个文件是经验值——在这个大小下删除一个空文件的代价可以忽略OS 的 unlink 操作是 O(1)。反模式三不限制消费速度错误做法网络恢复后瞬间把所有缓存消息发往云端。为什么错积压了 3 天、26GB 的消息如果无限制补传云端瞬间被几万 TPS 打满——其他正常节点的数据反而进不去了。正确做法TPS 限速消费默认 100 TPS在尽快补传和不给云端压力之间平衡。六、小结离线缓存是整个可靠性体系的最后一道防线数据到达 MessageHub │ ├─ 流控有令牌 → 正常发送 │ │ │ └─ 发送失败 → 写离线缓存 │ └─ 流控无令牌 → LOW 丢弃 → MEDIUM/HIGH 写离线缓存 │ ▼ 离线缓存队列 ├─ 写入异步 channel → 队尾文件追加 ├─ 淘汰FIFO 删最旧消息 ├─ 消费TPS 限速补传 ├─ 通知70%/60% 上下限滞回 └─ 老化超期消息自动清理从第 14 篇到第 17 篇我们完成了数据面最核心的可靠性设计。下一篇我们回到应用层——设备管理的全链路从设备认证到子设备拓扑到设备影子。本文是《边缘平台架构沉思录Go 架构推演与工程决策》系列的第 17 篇。