做工业数据采集的朋友尤其是做野外场站、市政管网、偏远厂区项目的大概率都踩过断网丢数据的坑。现场大多是4G/5G专网赶上刮风下雨、基站检修、甚至周边施工挖断光缆断网几十分钟到几小时都是常事。传统的直连上传模式断网期间的数据直接就丢了事后工艺分析、告警追溯、产量报表全对不上甲方追责下来背锅的还是我们做系统的。很多人一开始对付这个问题图省事就在程序里加个内存队列数据先存队列里网好了再发。结果遇上设备断电、程序异常崩溃内存里的数据直接清零等于白做。还有的直接写TXT/CSV文本时间长了文件越攒越大读写慢还容易损坏续传的时候要么重复传、要么漏数据乱七八糟根本没法用。这几年做了十几个边缘采集类项目从油田计量站到市政管网监测点网络条件一个比一个差踩遍了各种缓存方案的坑最后定型了一套基于SQLite的本地持久化缓存断点续传方案。断网、掉电、程序重启都不丢数据续传不重不漏还能控制带宽不挤占生产业务到现在现场稳定跑了两年多没出过一次数据丢失的问题。今天把完整的实现思路、核心代码和现场踩坑经验整理出来都是实际项目里验证过的拿来就能用。一、先讲清楚工业场景的断网续传到底要满足什么要求普通互联网应用的离线缓存放到工业采集场景基本都会水土不服。核心原因就在于工业场景对数据可靠性、时序性、采集稳定性的要求远高于普通场景总结下来有六个硬性要求数据零丢失这是底线。断网、掉电、程序异常退出缓存的数据都不能丢。纯内存方案直接排除一断电就清零根本扛不住工业现场的复杂工况。时序一致性上传的数据必须和采集顺序严格一致不能乱序。工业数据大多是按时序存储的一旦数据顺序乱了趋势曲线、统计分析、告警判断全都会出错比丢数据还麻烦。续传准确性不能重复传也不能漏传。很多粗糙的方案续传的时候要么从头重传一遍导致平台数据重复要么传错了断点位置中间漏了一大段事后根本查不出来。采集无阻塞缓存的写入绝对不能影响正常的采集频率。不能因为写缓存慢了就导致采集周期漂移、数据间隔不准。永远记住采集是主业传输是副业不能让副业拖垮主业。磁盘空间可控不能无限缓存把磁盘写满。工业现场的边缘网关、工控机磁盘空间都不大几个G的数据库就能把系统拖垮。必须有容量限制、过期清理机制自动循环覆盖。异常容错能力缓存文件损坏、磁盘出坏道、写入异常要有兜底机制。不能因为缓存模块出问题就导致整个采集程序崩溃最坏情况也要保证采集能正常运行。二、整体架构三层解耦采集优先整套方案采用「采集层-缓存层-传输层」三层解耦架构每层独立线程运行互不阻塞核心原则是采集永远优先传输后置兜底。采集层负责与PLC、传感器、采集模块通信按固定周期采集数据。采集完成后只需要把数据写入缓存层就立即返回完全不关心当前网络状态保证采集周期精准。缓存层基于嵌入式数据库做本地持久化存储负责数据写入、状态管理、查询检索。所有数据落地磁盘掉电不丢是整个方案的核心底座。传输层负责网络状态检测、数据批量上传、断点管理、重试逻辑。网络正常时接近实时上传断网时自动暂停上传数据全部沉淀到本地网络恢复后自动从断点开始续传。三层之间通过数据状态流转协作采集只管写、传输只管读缓存层做中间解耦从架构上避免了传输问题影响采集。三、缓存层设计与工程实现1. 为什么选SQLite做持久化对比过很多存储方案最后还是SQLite最适合工业边缘场景嵌入式单文件不需要安装任何服务部署简单拷贝就能用适配Windows、Linux嵌入式系统。支持ACID事务写入中途掉电不会损坏文件数据一致性有保障这是文本文件比不了的。支持索引查询几十万条数据里检索待上传记录毫秒级返回性能远高于遍历文本。支持多线程并发读写采集和上传同时操作也不会乱不用自己写一堆锁逻辑。2. 数据表设计核心就一张缓存表用自增主键保证时序用状态字段管理上传流程CREATE TABLE IF NOT EXISTS data_cache ( id INTEGER PRIMARY KEY AUTOINCREMENT, -- 自增主键天然保证采集顺序 device_code TEXT NOT NULL, -- 设备编码 collect_time INTEGER NOT NULL, -- 采集时间戳毫秒 data_content TEXT NOT NULL, -- 采集数据内容JSON格式 upload_status INTEGER DEFAULT 0, -- 0:待上传 1:上传中 2:成功 3:失败 retry_count INTEGER DEFAULT 0, -- 重试次数 create_time INTEGER DEFAULT 0 -- 写入时间 ); -- 建索引提升查询速度 CREATE INDEX IF NOT EXISTS idx_upload_status ON data_cache(upload_status); CREATE INDEX IF NOT EXISTS idx_collect_time ON data_cache(collect_time);这里有个关键设计用自增id作为时序基准。id是按写入顺序自增的和采集顺序完全一致后续按id从小到大上传天然保证了时序不乱不用额外做排序处理。3. 缓存读写核心实现封装一个独立的缓存服务类对外只暴露写入、查询、更新状态三个核心方法内部用事务保证数据一致性。/// summary /// 本地数据缓存服务 /// /summary public class LocalCacheService { private readonly string _dbPath; private readonly object _lockObj new object(); public LocalCacheService(string dbPath) { _dbPath dbPath; InitDatabase(); } private void InitDatabase() { using var conn new SQLiteConnection($Data Source{_dbPath};); conn.Open(); // 执行建表SQL省略非核心代码 } /// summary /// 写入采集数据 /// /summary public bool AddData(string deviceCode, long collectTime, string dataContent) { lock (_lockObj) { using var conn new SQLiteConnection($Data Source{_dbPath};); conn.Open(); using var trans conn.BeginTransaction(); try { string sql INSERT INTO data_cache (device_code, collect_time, data_content, upload_status, create_time) VALUES (deviceCode, collectTime, dataContent, 0, createTime); using var cmd new SQLiteCommand(sql, conn, trans); cmd.Parameters.AddWithValue(deviceCode, deviceCode); cmd.Parameters.AddWithValue(collectTime, collectTime); cmd.Parameters.AddWithValue(dataContent, dataContent); cmd.Parameters.AddWithValue(createTime, DateTimeOffset.Now.ToUnixTimeMilliseconds()); cmd.ExecuteNonQuery(); trans.Commit(); return true; } catch { trans.Rollback(); return false; } } } /// summary /// 获取待上传数据按id升序保证时序 /// /summary public ListCacheDataItem GetWaitingUploadList(int batchSize 100) { lock (_lockObj) { var result new ListCacheDataItem(); using var conn new SQLiteConnection($Data Source{_dbPath};); conn.Open(); string sql SELECT id, device_code, collect_time, data_content FROM data_cache WHERE upload_status IN (0, 3) ORDER BY id ASC LIMIT batchSize; using var cmd new SQLiteCommand(sql, conn); cmd.Parameters.AddWithValue(batchSize, batchSize); using var reader cmd.ExecuteReader(); while (reader.Read()) { result.Add(new CacheDataItem { Id reader.GetInt64(0), DeviceCode reader.GetString(1), CollectTime reader.GetInt64(2), DataContent reader.GetString(3) }); } return result; } } /// summary /// 批量标记上传成功 /// /summary public void MarkSuccess(long maxId) { lock (_lockObj) { using var conn new SQLiteConnection($Data Source{_dbPath};); conn.Open(); string sql UPDATE data_cache SET upload_status 2 WHERE id maxId AND upload_status ! 2; using var cmd new SQLiteCommand(sql, conn); cmd.Parameters.AddWithValue(maxId, maxId); cmd.ExecuteNonQuery(); } } /// summary /// 标记上传失败增加重试次数 /// /summary public void MarkFailed(long startId, long endId) { lock (_lockObj) { using var conn new SQLiteConnection($Data Source{_dbPath};); conn.Open(); string sql UPDATE data_cache SET upload_status 3, retry_count retry_count 1 WHERE id BETWEEN startId AND endId; using var cmd new SQLiteCommand(sql, conn); cmd.Parameters.AddWithValue(startId, startId); cmd.Parameters.AddWithValue(endId, endId); cmd.ExecuteNonQuery(); } } }这里加了锁机制保证线程安全采集线程和上传线程同时操作也不会出问题。写入用事务要么成功要么失败不会出现半条数据的情况掉电也不怕。四、传输层断网检测与断点续传逻辑传输层是整个方案的调度核心负责判断网络状态、控制上传节奏、处理断点续传。最关键的是不能网络稍微抖动就进入断网模式也不能断网了还一直死循环重试。1. 网络状态判断机制采用「连续失败判定周期性探活」的策略避免网络抖动导致频繁切换正常传输状态下连续3次批量上传失败才判定为断网进入缓存模式。断网状态下每30秒发起一次探活请求发送少量测试数据。连续2次探活成功判定网络恢复进入续传模式开始补传历史数据。历史数据补传完成后切换回实时传输模式。2. 断点续传核心逻辑续传的核心是按id顺序批量处理成功一批标记一批断点就是最后一次成功的最大id下次从这个id之后继续天然保证不重不漏。完整流程从缓存表按id升序拉取一批待上传数据默认100条。打包压缩后调用服务端批量上传接口。接口返回成功的最大id将该id之前的数据全部标记为成功。如果接口调用失败记录失败次数等待后重试。循环拉取下一批直到没有待上传数据续传完成。核心代码实现/// summary /// 数据上传管理器 /// /summary public class UploadManager { private readonly LocalCacheService _cacheService; private readonly string _serverUrl; private readonly Thread _uploadThread; private volatile bool _isRunning; private volatile NetWorkStatus _networkStatus NetWorkStatus.Normal; private int _连续失败次数 0; private const int 断网阈值 3; public UploadManager(LocalCacheService cacheService, string serverUrl) { _cacheService cacheService; _serverUrl serverUrl; _uploadThread new Thread(UploadLoop); _uploadThread.IsBackground true; } public void Start() { _isRunning true; _uploadThread.Start(); } private void UploadLoop() { while (_isRunning) { try { if (_networkStatus NetWorkStatus.Disconnected) { // 断网状态30秒探活一次 Thread.Sleep(30000); if (CheckNetworkAlive()) { _networkStatus NetWorkStatus.Normal; _连续失败次数 0; } continue; } // 获取待上传数据 var list _cacheService.GetWaitingUploadList(100); if (list.Count 0) { Thread.Sleep(1000); continue; } // 批量上传 bool success BatchUpload(list, out long maxSuccessId); if (success) { _cacheService.MarkSuccess(maxSuccessId); _连续失败次数 0; } else { _连续失败次数; if (_连续失败次数 断网阈值) { _networkStatus NetWorkStatus.Disconnected; } // 失败后等待重试 Thread.Sleep(5000); } } catch (Exception ex) { // 异常不退出循环保证服务不中断 Thread.Sleep(10000); } } } private bool BatchUpload(ListCacheDataItem list, out long maxSuccessId) { maxSuccessId 0; try { // 打包数据、压缩、调用接口省略HTTP调用细节 var requestData new { data list }; // 调用服务端接口 var response HttpPost(_serverUrl /api/data/batchUpload, requestData); if (response.Success) { maxSuccessId response.MaxId; return true; } return false; } catch { return false; } } private bool CheckNetworkAlive() { // 简单的接口探活或者ping测试 try { var response HttpGet(_serverUrl /api/health); return response.Success; } catch { return false; } } }五、关键工程化优化这些细节决定现场稳不稳核心逻辑写完只是能用要想在工业现场稳定跑还有很多细节要处理都是踩坑踩出来的经验。1. 磁盘空间管理环形缓存自动清理一定要设置缓存上限不然跑半年磁盘就满了。我们采用「按容量按时间」双重清理策略最多保留10万条已上传数据超过就删除最早的部分。已上传数据最多保留7天超过自动清理。每天凌晨定时执行清理任务避开白天生产高峰期。磁盘剩余空间低于10%时强制清理最早的已上传数据优先保证新数据能写入。2. 采集与缓存异步解耦采集频率高的场景比如每秒几十上百次不要在采集线程里直接写数据库不然数据库偶尔卡顿会导致采集周期漂移。可以加一级内存队列做缓冲采集线程只写内存队列单独开一个后台线程批量从队列里取数据写入数据库。既保证了采集的实时性又提升了数据库写入效率。注意内存队列不要攒太多最多攒1秒的数据避免掉电丢失过多。对可靠性要求极高的场景还是逐条事务写入。3. 批量压缩传输工业现场带宽普遍不高批量上传的时候一定要做压缩。100条JSON数据用GZIP压缩后体积只有原来的1/5~1/3传输效率提升非常明显弱网下成功率也高很多。4. 幂等性保证网络超时的时候服务端可能已经处理成功了但终端没收到响应就会认为失败重发导致数据重复。服务端一定要做幂等校验用「设备编码采集时间戳」作为唯一键重复的数据直接跳过返回成功即可。终端不用改服务端兜底是最稳妥的方案。5. 数据库损坏兜底SQLite虽然稳定但也有可能因为异常断电、磁盘坏道损坏。可以做双备份机制每天凌晨自动备份一次完整的数据库文件。程序启动时检测数据库完整性如果损坏自动从最近的备份恢复。如果备份也损坏就自动重建新的数据库牺牲历史数据保证采集能正常运行。六、服务端配合要点终端侧做的再好服务端不配合也白搭。服务端只要做好三点就能完美适配这套方案提供批量上传接口不要只做单条上传批量处理性能差几十倍。接口接收数据数组批量写入时序库。返回成功最大ID接口响应里返回本次处理成功的最大id终端根据这个id批量更新状态不用每条都确认效率极高。实现幂等去重根据设备编码采集时间去重避免重复数据。七、现场踩过的那些坑别用内存队列当主缓存早年图省事用过ConcurrentQueue存内存结果现场设备半夜断电重启断网4小时的数据全没了连夜跑现场补数据教训惨痛。内存只能做缓冲绝对不能当最终存储。别用文本文件存缓存也试过写CSV文件时间长了文件几个G查待上传数据要遍历整个文件慢得要死。而且并发写入容易乱异常退出经常损坏文件数据都读不出来。断网判断别太灵敏最开始设成失败1次就判定断网结果网络稍微抖动一下就进入缓存模式频繁切换反而导致很多不必要的缓存。改成连续失败3次才判定就稳定多了。别忘了清理已上传数据有个项目没做清理跑了半年缓存数据库涨到12G设备磁盘满了采集程序直接崩了。一定要定期清理设置容量红线。不要在采集线程里写数据库一开始图简单采集线程直接写数据库遇上数据库锁等待采集周期从1秒变成5秒数据都不准了。采集和存储一定要分层解耦。最后说几句工业数据采集稳定性永远是第一位的。网络条件差是客观环境很多时候我们没法改变能做的就是在软件层面做好兜底保证数据不丢。这套方案本质上就是用本地持久化做缓冲把网络的波动和采集层完全隔离开。不管网络断多久只要本地磁盘够数据就不会丢网好了自动同步上去业务层完全无感知。方案不复杂用的都是成熟技术但细节很多每一个坑都是现场跑出来的经验。大家落地的时候不用完全照搬根据自己的采集频率、设备性能、网络条件调整参数就行核心的持久化、有序性、幂等、磁盘管理这几点一定要做好这是不丢数据的根基。