记不清有多少次了我蹲在现场机柜旁边看着网关的上行链路微微指示灯一闪然后数据就断了。上位机弹出一堆红色报警归结起来就一句话历史数据断了丢了一段。工控网关接PLC、接智能仪表、接电表什么都好就是网络不稳定的时候你得想办法让数据先别丢、后补发。这个场景里我踩了几次坑之后越来越觉得 Outbox 模式是个好用的套路。干过厂区信息化或者边缘采集的老哥们应该都有这种体会网关和上位机/云平台之间只要到了现场网络就没法拿办公室的标准来衡量。可能是一条光纤被挖断了也可能是某个交换机晚上被断电甚至就是运营商基站抖动一下一两分钟就恢复了。可就是这一两分钟产线上的几十个点位数据全乱了。传统直采直发根本顶不住内存队列也扛不住真正能落地的是让网关像邮局一样——先把信写在本子上存好再慢慢往外寄。这就是工控网关里的 Outbox 模式也是我这篇文章想认真聊聊的东西。1. 我在现场遇到的那些断网数据事故1.1 产线一个抖动上位机少了半小时数据最早我自己做网关采集的时候思路没什么争议轮询PLC寄存器拿到数据就通过MQTT封包往服务器推。这套逻辑在小规模测试环境里跑得风生水起延时低、占用小、代码还简单。真正到了客户现场第一个问题就来了。客户车间里有条老产线一台设备每隔几秒就有几十个变量在变化网关按500毫秒周期采集数据到上位机以后用来做产量统计和质量追溯。那天上午车间修光纤的时候把网关所在交换机也给掐了前前后后不到四十分钟。结果是什么采集到的数据全都堆在内存链表里内存还好说但是MQTT连接一断客户端库不断重连。链路恢复以后上位机倒是把延迟积压的数据收走了可中间有一段正好跨过断网窗口的记录因为网关重启了一次——现场工人不知道怎么弄把网关电源给拍了。数据在内存里没来得及刷进本地存储直接没了。事后领导只问一句为什么少了一段我愣是解释不清楚。这个事故给了我一个非常深刻的教训网关跟服务器之间的链路永远不能假设是可靠的。1.2 内存队列为什么帮不了你可能你会想那我给网关加一个内存队列缓存断网的时候数据往队列里塞恢复后慢慢发不就行了确实这比直发好很多很多工业网关的固件也是这样处理的。但内存队列有一个致命问题它扛不住进程崩溃和断电。嵌入式Linux上跑的应用哪怕你用环形队列、用共享内存只要断电一瞬间RAM里的所有缓存直接清零。而工业现场的断电有时候恰恰不在你能控制的范围——电网晃一下、空开跳了、操作工拔错插座这些我都见过。更隐蔽的问题是内存队列没有持久化语义你就很难回答哪些数据我已经发给服务器了、哪些还没发。哪怕你在内存里维护一个已发送偏移量服务器有没有真实收到你还是不知道。一旦服务器那边处理到一半崩了或者MQTT断开前最后一包报文只发了一半你就得面对消息可能丢、可能重复、可能乱序的三重窘境。所以真正能兜底的方案必须把待发送数据落到非易失存储上并且有明确的发送状态追踪机制。这就是为什么我把目光转向了 Outbox 模式。1.3 什么才算工业级的数据不丢工业现场谈不丢数据首先要定义一下粒度。对于点位采集来说最基本的保证是网关采集到的一条记录一旦进入本地持久化存储就必须一直留到它被服务器确认收到为止。断网十年也好断电重启一百次也好这条记录不能丢。如果容量有限导致写满那也要遵循先进先出的淘汰策略而不是毫无预警地在新数据写入时被覆盖。这一点和数据库领域的 Durability 很像。而 Outbox 模式恰恰能给你这个保证因为它把要发送的数据、发送状态、发送时间、重试次数全部以记录形式存在本地数据库里。就算网关因为各种原因宕机重启后一扫描 Outbox 表该发的数据一条都不会少。它不是通过什么高深算法解决数据一致性的它靠的是先落盘再发送记状态做确认这十二个字。理解了这十二个字你就理解了这个模式的工程内核。2. Outbox 模式到底是怎么运作的2.1 一次事务里同时干完两件事在微服务架构中Transactional Outbox 的核心是业务操作和发消息操作在同一个数据库事务里完成。比如订单服务在 MySQL 里插入一条订单记录同时往 outbox 表插入一条订单已创建待发送记录。两个写入在同一个事务中要么都成功要么都失败。这样就避免了业务数据保存了但消息没发出去导致下游系统感知不到的不一致问题。工控网关里的思路是一模一样的。网关每轮采集到一批点位数据不是直接往 MQTT broker 发而是先在自己的本地数据库里开启一个事务把点位数据写入数据表同时把对应的待发送任务插入 outbox 表。事务提交两块数据都稳稳落盘。之后由另一个独立的发送模块去扫描 outbox 表把这些记录一条条发出去发成功的再更新状态为已确认或者直接删除。数据先落库发送后置两个动作之间不做强同步耦合——这就是和先发送后补偿思路最本质的区别。2.2 发送模块的三步闭环扫描、发送、确认Outbox 表在网关里的角色可以理解成一个待办的寄件列表。发送模块的循环逻辑就三个步骤第一步扫描 outbox 表找出所有状态为 pending 的记录按时间排序第二步逐条或批量打包通过 MQTT/Modbus TCP 把数据发出去第三步接收服务器侧返回的确认可能是 MQTT 的 ACK、消息 ID 回执或者应用层自定义的确认报文确认成功后把 outbox 记录标记为 sent或者直接从表中删除。如果发送失败比如超时没有确认记录状态不变重试次数加一等待下一次扫描。如果网络一直没恢复扫描任务会因为发送失败不断重试但记录始终躺在表里不会凭空消失。断网期间网关可以放心大胆地把新数据继续插入 outbox 表网络恢复后发送模块会按时间顺序把积压的数据一条条补发出去。整个过程不需要人工干预网关就像一台自动记账并持续结算的邮局。2.3 不是简单的重传是最终一致性的本地化实践很多人第一次听说 Outbox 会觉得这不就是断点续传加重试机制吗其实两者有细微但很重要的区别。断点续传通常针对的是一个文件或一条数据流重试机制面向的是单条消息的发送失败而 Outbox 模式的着眼点是让产生数据和发送数据这两个动作在一致性层面解耦并把解耦的边界建立在可靠的本地存储之上。我之前在网关里做过一套专门的重发逻辑出问题后靠 TCP 连接状态判断哪些数据没发送。结果发现 TCP 层 send 成功并不等于对端应用真的处理了这条数据很可能 broker 收下了但转发到上位机的时候挂了。Outbox 的模式就会逼着你把确认层级上升到应用层服务器处理完数据、落库之后再给网关回一条确认消息网关收到这条确认才把 outbox 记录删掉。这个语义和 MQTT QoS 2、或者 Modbus TCP 的响应报文都是能对上的。说白了Outbox 最核心的贡献不是重试而是把什么算发成功定义清楚了。3. 手把手在工控网关上实现 Outbox3.1 网关里的 outbox 表字段应该这样设计在嵌入式 Linux 网关里最常用的本地数据库无非是 SQLite 或者轻量的 LevelDB/RocksDB。SQLite 因为支持完整事务、SQL 操作方便而且单文件存储好备份比较适合做 Outbox 的载体。我个人推荐表结构长这样CREATE TABLE outbox ( id INTEGER PRIMARY KEY AUTOINCREMENT, topic TEXT NOT NULL, -- 发送主题例如 factory/line1/machine3 payload BLOB NOT NULL, -- 序列化后的点位数据可能是 JSON 或二进制 status INTEGER DEFAULT 0, -- 0pending, 1sending, 2sent, 3dead retry_count INTEGER DEFAULT 0, -- 已重试次数 next_retry_time INTEGER DEFAULT 0, -- 下次扫描时间戳毫秒级 create_time INTEGER NOT NULL, -- 创建时间戳 send_time INTEGER -- 实际发送成功时间 ); CREATE INDEX idx_outbox_status_time ON outbox(status, next_retry_time);几个字段设计的关键点我单独说一下。payload 用 BLOB 而不是 TEXT是因为工业现场的数据可能是 Modbus 寄存器拼接出来的二进制结构转到 JSON 再入库反而费 CPU直接存原始字节更通用。status 字段建议至少保留 pending、sending、sent、dead 四个状态不要偷懒只用一个 pending 加一个删除。因为在发送过程中如果进程刚好崩溃状态还在 sending 的记录重启后需要能被识别出来重新回到 pending。retry_count 和 next_retry_time 是给重试策略用的避免网络不好时高频重试打爆 CPU 和数据库。那个联合索引很重要。扫描发送任务的时候只会查 status 和 next_retry_time 这两个字段不建索引的话数据量到几十万条扫描会明显变慢。工业网关的存储空间一般不大但点位一多、采集周期一短积压个几十万条真的很常见。3.2 写入侧一个事务锁住数据表和 outbox 表写入侧的逻辑要保证原子性这是整个 Outbox 模式的关键命门。轮询线程读到 PLC 的数据以后不能先写数据表、再插 outbox因为这两步之间如果断电数据表有了而 outbox 没有服务器就不会收到这段数据。正确做法是BEGIN TRANSACTION; INSERT INTO tag_history(tag_id, timestamp, value) VALUES (?, ?, ?); INSERT INTO outbox(topic, payload, status, create_time, next_retry_time) VALUES (factory/line1/tags, ?, 0, ?, 0); COMMIT;这段 SQL 看上去平淡无奇但它在工程上的意义非常重大。它把我要保存这条数据和我要发送这条数据这两个意图绑在了一个事务里要么一起成功要么一起失败。实际编码时要注意SQLite 默认的 journal 模式最好是 WAL因为 WAL 模式在意外断电后的恢复能力更强而且读写并发好一些。另外不要把大事务搞得太长——每轮采集最多也就是写入几十个点位和几十条 outbox 记录事务执行完立刻提交否则会导致写入延迟累积影响后续采样周期。我在一个项目里遇到采集周期 100ms、一个事务里塞了上千条记录的情况结果网关 CPU 直接飘到 70% 以上。拆小批量以后CPU 降到了 15% 左右。还有一种情况要特别处理如果网关采集到的数据本身就不需要发送只是本地存储那就不需要写 outbox。Outbox 表里只放需要上行同步的数据。否则表膨胀会特别快造成无意义的磁盘和扫描压力。3.3 发送侧循环定时扫描加指数退避重试发送模块我建议单独起一个线程避免阻塞采集主线程。它做的事情就是轮询扫描 outbox 表把 pending 状态的数据取出来发送。扫描策略可以设计成这样每隔一定时间比如 1 秒扫描一次每次取前 N 条按 create_time 排序批量发送。发送流程我给出一个简洁的 C 语言伪代码void *outbox_sender_thread(void *arg) { while (1) { // 1. 取待发送记录按时间正序 rows db_query( SELECT id, topic, payload FROM outbox WHERE status IN (0,1) AND next_retry_time ? ORDER BY create_time ASC LIMIT 200, now_ms()); if (rows_empty) { sleep(1); // 没有任务就歇一秒 continue; } foreach (row in rows) { // 2. 标记为 sending防止重复取出 db_exec(UPDATE outbox SET status1 WHERE id?, row.id); // 3. 构造 MQTT 报文发送并等待应用层确认 result mqtt_publish_and_wait_ack(row.topic, row.payload, 5000); if (result ACK_OK) { // 4. 服务器确认成功直接删除记录 db_exec(DELETE FROM outbox WHERE id?, row.id); } else { // 5. 发送失败更新重试次数和下次扫描时间 db_exec( UPDATE outbox SET status0, retry_countretry_count1, next_retry_time? WHERE id?, now_ms() retry_backoff(retry_count), row.id); } } } }重试时间的计算我推荐指数退避加一点抖动。比如第一次重试等 5 秒第二次 10 秒第三次 20 秒上限 5 分钟。不要用固定间隔因为网络恢复往往有一个过程高频重试只会白白占用带宽和 CPU。抖动是为了防止多条记录在同一时刻同时重试造成流量尖峰。还有一个小细节status 标记为 1sending以后如果应用在等待 ACK 的 5 秒内崩溃了这条记录会一直停留在 sending 状态。所以在扫描条件里我加了 status IN (0,1)同时把 next_retry_time 在过去时间点的 sending 记录重新捞出来处理。宁可重复发送也不能让它卡死在 sending 里永远发不出去。这就是网络通信里的 at-least-once 语义。3.4 服务器侧要配合做幂等因为发送侧是 at-least-once 语义服务器在接收这些数据的时候必须做幂等处理。最直接的办法是网关每一条 outbox 记录生成一个全局唯一的消息 ID服务器收到数据后以这个消息 ID 作为主键或唯一索引去重。哪怕同一条数据因为网关崩溃恢复被重发了三遍服务器最终存储的数据也只有一份。具体到实现上可以在 outbox 表里加一个 msg_id 字段内容用 UUID 或时间戳加随机数生成。MQTT 消息的 payload 里包含这个 msg_id服务器解析 payload 后检查 Redis 或者数据库里有没有处理过。没有就正常入库有就回一个 ACK 但不重复写库。这一步如果省了生产环境大概率会碰到重复数据。不要觉得概率低网关断网重连那一刻最后一两条消息重发是非常常见的。3.5 积压太多怎么办限量发送和优先级处理断网时间一长outbox 表里积压的数据可能达到几十万条。这时如果网络恢复发送模块如果一次性把所有数据全捞出来网关内存会瞬间被打爆。所以扫描的时候必须限制每轮取出的条数我的经验是一轮不超过 200 条或者不超过 1MB 的 payload。发完这一批再扫下一批。这样即使有大量积压也只是发送耗时变长系统不会崩溃。优先级的问题也要考虑。有些现场数据是实时报警类有些是历史记录类。如果 mixed 在一个 outbox 表里断网恢复后按时间顺序补发历史数据报警信息会被延迟很久。我常用的做法是拆两个表或者在同一张表里加一个 priority 字段。报警类记录的 priority 设为 1扫描发送时先按 priority、再按 create_time 排序。这样既保证历史数据不丢又不耽误实时告警的及时性。4. 常见问题与排查技巧实录4.1 数据重复到达服务器库里出现了双份记录这是我在项目里被客户怼得最多的问题。第一次出现的时候我看日志发现 MQTT 客户端在重连成功那一刻之前发送超时但 broker 实际上已经处理掉的消息被网关重发了一遍。服务器没做幂等结果库里多了一批重复记录。排查思路分两层。先看网关侧是否真的产生了重复的 outbox 记录可以通过 msg_id 的 unique 约束来验证。再看服务器侧有没有做消息 ID 去重。我发现最有效的修法就是在服务器上给 msg_id 建唯一索引并采用先查后插或insert ignore策略。网关侧则要把发送确认机制落实到位只有服务器显式回 ACK 的记录才能删除TCP 发送成功不算数。4.2 断网恢复后历史数据补发顺序乱掉了数据乱序通常不是 outbox 表本身的问题而是发送线程在并发发送时没有控制好序号。如果你开了多个发送线程同时扫同一个 outbox 表不同线程取到的时间靠前的记录可能和后取到的记录交错发送服务器落库顺序自然就乱了。对大多数工控场景来说同一台设备的数据如果乱序曲线图就会出现毛刺和回跳。解决思路有两个。一是发送模块只开单线程严格按 create_time 顺序逐条发。二是如果对吞吐量有要求可以按设备 ID 做哈希分桶每个桶一个线程保证同一设备的记录严格有序。第二个方案对采集周期短、点位数量大的场景更实用。4.3 SQLite 在掉电瞬间丢记录的坑SQLite 在嵌入式设备上有一个经典坑如果建库时没有设置好同步模式掉电可能导致最近几条已提交事务丢失。默认的 synchronousFULL 时SQLite 在提交事务时要等数据真正落到磁盘才返回掉电安全最好但性能会慢一些而 synchronousNORMAL 在 WAL 模式下性能好很多但断电时可能丢失最后几个事务。工业网关掉电是常态我的建议是不要为了性能把 synchronous 降到 OFF尽量保留 FULL 或者至少 NORMAL 加 WAL。而且网关最好配一个小的 UPS 或者带掉电检测功能在检测到电压跌落时程序先停止新一轮采集把 outbox 表的数据紧急刷盘然后进入安全的关机流程。没有这个硬件配合任何软件方案都没法保证绝对的掉电零丢失。4.4 outbox 表太大磁盘被撑满怎么处理极端情况下如果网络连续断了几天网关本身存储空间又只有几百 MBoutbox 表是有可能撑满的。磁盘满了以后采集线程往数据库写入也会失败导致新数据都无法落库这会让问题雪上加霜。所以一定要给 outbox 表设置容量上限和淘汰策略。我用过一个比较实用的组合策略给 outbox 表按时间分区比如一天一个分区同时设置最大保留时间和最大行数。当超出阈值时先把超过 7 天的记录导出一个 CSV 备份文件再从表里删除。这样既不会丢历史数据也不会让新数据写不进去。读取侧发送模块要对表被清理的情况做防护如果记录没了就跳过继续发后面的别因为一条删除掉的记录卡死整个发送循环。4.5 云平台找不到网关最后几分钟的数据还有一种常见情况是网络是通的数据也在实时发送但是服务器侧不知道什么原因漏收了最后几分钟的数据。这种问题最隐蔽因为网关 outbox 表自己可能已经清空了——它收到了 ACK 就删除了记录。等到服务器那边发现数据有洞日志早就找不到了。排查我一般分三步。第一步确认服务器 ACK 报文的生成时间对比数据写入时间看是不是网络延迟导致 ACK 回得特别晚第二步检查网关的本地历史数据表里有没有这几分钟的数据如果网关侧落库了那就是发送或者服务器存储环节的问题第三步在服务器入库接口加 Trace ID 全链路日志把从 MQTT 收到报文到落库成功的每一步时间戳打出来定位到底哪一步丢了。最后提醒一下给 outbox 加上记录清理延迟策略也有帮助比如 ACK 成功后不要立刻删除记录而是先标记为 sent 状态保留 24 小时再异步清理。这样服务器如果哪天发现数据有洞还能倒查网关日志找原因。5. 我用这套方案重新改造网关后的一些体会改造完成之后我最直观的感受是再也不用盯着网络抖动看监控了。以前网络一抖我就紧张现在只要 outbox 表里没积压我基本不关心链路什么时候恢复。数据要么躺在本地的 outbox 表里要么已经发给服务器并被确认不会出现第三种我不知道它在哪的状态。这种确定感在工业现场比什么都值钱。最后再分享一个具体的小技巧。如果你用 SQLite 做 outbox 存储一定要定时执行 VACUUM 或者用自动 checkpoint 控制 WAL 文件大小。我遇到过几次 WAL 文件涨到好几个 GB导致磁盘告警的情况。最开始没注意后来在采集线程每次提交事务后加了一个 WAL 大小的检查超过阈值就主动执行一次 checkpoint问题就再也没出现过。这是个很不起眼的细节但对网关这种常年不断电的设备来说存储层的健康直接决定了 Outbox 模式能稳定跑多久。 Outbox模式在工控网关中的实践：断网数据不丢的核心方案
记不清有多少次了我蹲在现场机柜旁边看着网关的上行链路微微指示灯一闪然后数据就断了。上位机弹出一堆红色报警归结起来就一句话历史数据断了丢了一段。工控网关接PLC、接智能仪表、接电表什么都好就是网络不稳定的时候你得想办法让数据先别丢、后补发。这个场景里我踩了几次坑之后越来越觉得 Outbox 模式是个好用的套路。干过厂区信息化或者边缘采集的老哥们应该都有这种体会网关和上位机/云平台之间只要到了现场网络就没法拿办公室的标准来衡量。可能是一条光纤被挖断了也可能是某个交换机晚上被断电甚至就是运营商基站抖动一下一两分钟就恢复了。可就是这一两分钟产线上的几十个点位数据全乱了。传统直采直发根本顶不住内存队列也扛不住真正能落地的是让网关像邮局一样——先把信写在本子上存好再慢慢往外寄。这就是工控网关里的 Outbox 模式也是我这篇文章想认真聊聊的东西。1. 我在现场遇到的那些断网数据事故1.1 产线一个抖动上位机少了半小时数据最早我自己做网关采集的时候思路没什么争议轮询PLC寄存器拿到数据就通过MQTT封包往服务器推。这套逻辑在小规模测试环境里跑得风生水起延时低、占用小、代码还简单。真正到了客户现场第一个问题就来了。客户车间里有条老产线一台设备每隔几秒就有几十个变量在变化网关按500毫秒周期采集数据到上位机以后用来做产量统计和质量追溯。那天上午车间修光纤的时候把网关所在交换机也给掐了前前后后不到四十分钟。结果是什么采集到的数据全都堆在内存链表里内存还好说但是MQTT连接一断客户端库不断重连。链路恢复以后上位机倒是把延迟积压的数据收走了可中间有一段正好跨过断网窗口的记录因为网关重启了一次——现场工人不知道怎么弄把网关电源给拍了。数据在内存里没来得及刷进本地存储直接没了。事后领导只问一句为什么少了一段我愣是解释不清楚。这个事故给了我一个非常深刻的教训网关跟服务器之间的链路永远不能假设是可靠的。1.2 内存队列为什么帮不了你可能你会想那我给网关加一个内存队列缓存断网的时候数据往队列里塞恢复后慢慢发不就行了确实这比直发好很多很多工业网关的固件也是这样处理的。但内存队列有一个致命问题它扛不住进程崩溃和断电。嵌入式Linux上跑的应用哪怕你用环形队列、用共享内存只要断电一瞬间RAM里的所有缓存直接清零。而工业现场的断电有时候恰恰不在你能控制的范围——电网晃一下、空开跳了、操作工拔错插座这些我都见过。更隐蔽的问题是内存队列没有持久化语义你就很难回答哪些数据我已经发给服务器了、哪些还没发。哪怕你在内存里维护一个已发送偏移量服务器有没有真实收到你还是不知道。一旦服务器那边处理到一半崩了或者MQTT断开前最后一包报文只发了一半你就得面对消息可能丢、可能重复、可能乱序的三重窘境。所以真正能兜底的方案必须把待发送数据落到非易失存储上并且有明确的发送状态追踪机制。这就是为什么我把目光转向了 Outbox 模式。1.3 什么才算工业级的数据不丢工业现场谈不丢数据首先要定义一下粒度。对于点位采集来说最基本的保证是网关采集到的一条记录一旦进入本地持久化存储就必须一直留到它被服务器确认收到为止。断网十年也好断电重启一百次也好这条记录不能丢。如果容量有限导致写满那也要遵循先进先出的淘汰策略而不是毫无预警地在新数据写入时被覆盖。这一点和数据库领域的 Durability 很像。而 Outbox 模式恰恰能给你这个保证因为它把要发送的数据、发送状态、发送时间、重试次数全部以记录形式存在本地数据库里。就算网关因为各种原因宕机重启后一扫描 Outbox 表该发的数据一条都不会少。它不是通过什么高深算法解决数据一致性的它靠的是先落盘再发送记状态做确认这十二个字。理解了这十二个字你就理解了这个模式的工程内核。2. Outbox 模式到底是怎么运作的2.1 一次事务里同时干完两件事在微服务架构中Transactional Outbox 的核心是业务操作和发消息操作在同一个数据库事务里完成。比如订单服务在 MySQL 里插入一条订单记录同时往 outbox 表插入一条订单已创建待发送记录。两个写入在同一个事务中要么都成功要么都失败。这样就避免了业务数据保存了但消息没发出去导致下游系统感知不到的不一致问题。工控网关里的思路是一模一样的。网关每轮采集到一批点位数据不是直接往 MQTT broker 发而是先在自己的本地数据库里开启一个事务把点位数据写入数据表同时把对应的待发送任务插入 outbox 表。事务提交两块数据都稳稳落盘。之后由另一个独立的发送模块去扫描 outbox 表把这些记录一条条发出去发成功的再更新状态为已确认或者直接删除。数据先落库发送后置两个动作之间不做强同步耦合——这就是和先发送后补偿思路最本质的区别。2.2 发送模块的三步闭环扫描、发送、确认Outbox 表在网关里的角色可以理解成一个待办的寄件列表。发送模块的循环逻辑就三个步骤第一步扫描 outbox 表找出所有状态为 pending 的记录按时间排序第二步逐条或批量打包通过 MQTT/Modbus TCP 把数据发出去第三步接收服务器侧返回的确认可能是 MQTT 的 ACK、消息 ID 回执或者应用层自定义的确认报文确认成功后把 outbox 记录标记为 sent或者直接从表中删除。如果发送失败比如超时没有确认记录状态不变重试次数加一等待下一次扫描。如果网络一直没恢复扫描任务会因为发送失败不断重试但记录始终躺在表里不会凭空消失。断网期间网关可以放心大胆地把新数据继续插入 outbox 表网络恢复后发送模块会按时间顺序把积压的数据一条条补发出去。整个过程不需要人工干预网关就像一台自动记账并持续结算的邮局。2.3 不是简单的重传是最终一致性的本地化实践很多人第一次听说 Outbox 会觉得这不就是断点续传加重试机制吗其实两者有细微但很重要的区别。断点续传通常针对的是一个文件或一条数据流重试机制面向的是单条消息的发送失败而 Outbox 模式的着眼点是让产生数据和发送数据这两个动作在一致性层面解耦并把解耦的边界建立在可靠的本地存储之上。我之前在网关里做过一套专门的重发逻辑出问题后靠 TCP 连接状态判断哪些数据没发送。结果发现 TCP 层 send 成功并不等于对端应用真的处理了这条数据很可能 broker 收下了但转发到上位机的时候挂了。Outbox 的模式就会逼着你把确认层级上升到应用层服务器处理完数据、落库之后再给网关回一条确认消息网关收到这条确认才把 outbox 记录删掉。这个语义和 MQTT QoS 2、或者 Modbus TCP 的响应报文都是能对上的。说白了Outbox 最核心的贡献不是重试而是把什么算发成功定义清楚了。3. 手把手在工控网关上实现 Outbox3.1 网关里的 outbox 表字段应该这样设计在嵌入式 Linux 网关里最常用的本地数据库无非是 SQLite 或者轻量的 LevelDB/RocksDB。SQLite 因为支持完整事务、SQL 操作方便而且单文件存储好备份比较适合做 Outbox 的载体。我个人推荐表结构长这样CREATE TABLE outbox ( id INTEGER PRIMARY KEY AUTOINCREMENT, topic TEXT NOT NULL, -- 发送主题例如 factory/line1/machine3 payload BLOB NOT NULL, -- 序列化后的点位数据可能是 JSON 或二进制 status INTEGER DEFAULT 0, -- 0pending, 1sending, 2sent, 3dead retry_count INTEGER DEFAULT 0, -- 已重试次数 next_retry_time INTEGER DEFAULT 0, -- 下次扫描时间戳毫秒级 create_time INTEGER NOT NULL, -- 创建时间戳 send_time INTEGER -- 实际发送成功时间 ); CREATE INDEX idx_outbox_status_time ON outbox(status, next_retry_time);几个字段设计的关键点我单独说一下。payload 用 BLOB 而不是 TEXT是因为工业现场的数据可能是 Modbus 寄存器拼接出来的二进制结构转到 JSON 再入库反而费 CPU直接存原始字节更通用。status 字段建议至少保留 pending、sending、sent、dead 四个状态不要偷懒只用一个 pending 加一个删除。因为在发送过程中如果进程刚好崩溃状态还在 sending 的记录重启后需要能被识别出来重新回到 pending。retry_count 和 next_retry_time 是给重试策略用的避免网络不好时高频重试打爆 CPU 和数据库。那个联合索引很重要。扫描发送任务的时候只会查 status 和 next_retry_time 这两个字段不建索引的话数据量到几十万条扫描会明显变慢。工业网关的存储空间一般不大但点位一多、采集周期一短积压个几十万条真的很常见。3.2 写入侧一个事务锁住数据表和 outbox 表写入侧的逻辑要保证原子性这是整个 Outbox 模式的关键命门。轮询线程读到 PLC 的数据以后不能先写数据表、再插 outbox因为这两步之间如果断电数据表有了而 outbox 没有服务器就不会收到这段数据。正确做法是BEGIN TRANSACTION; INSERT INTO tag_history(tag_id, timestamp, value) VALUES (?, ?, ?); INSERT INTO outbox(topic, payload, status, create_time, next_retry_time) VALUES (factory/line1/tags, ?, 0, ?, 0); COMMIT;这段 SQL 看上去平淡无奇但它在工程上的意义非常重大。它把我要保存这条数据和我要发送这条数据这两个意图绑在了一个事务里要么一起成功要么一起失败。实际编码时要注意SQLite 默认的 journal 模式最好是 WAL因为 WAL 模式在意外断电后的恢复能力更强而且读写并发好一些。另外不要把大事务搞得太长——每轮采集最多也就是写入几十个点位和几十条 outbox 记录事务执行完立刻提交否则会导致写入延迟累积影响后续采样周期。我在一个项目里遇到采集周期 100ms、一个事务里塞了上千条记录的情况结果网关 CPU 直接飘到 70% 以上。拆小批量以后CPU 降到了 15% 左右。还有一种情况要特别处理如果网关采集到的数据本身就不需要发送只是本地存储那就不需要写 outbox。Outbox 表里只放需要上行同步的数据。否则表膨胀会特别快造成无意义的磁盘和扫描压力。3.3 发送侧循环定时扫描加指数退避重试发送模块我建议单独起一个线程避免阻塞采集主线程。它做的事情就是轮询扫描 outbox 表把 pending 状态的数据取出来发送。扫描策略可以设计成这样每隔一定时间比如 1 秒扫描一次每次取前 N 条按 create_time 排序批量发送。发送流程我给出一个简洁的 C 语言伪代码void *outbox_sender_thread(void *arg) { while (1) { // 1. 取待发送记录按时间正序 rows db_query( SELECT id, topic, payload FROM outbox WHERE status IN (0,1) AND next_retry_time ? ORDER BY create_time ASC LIMIT 200, now_ms()); if (rows_empty) { sleep(1); // 没有任务就歇一秒 continue; } foreach (row in rows) { // 2. 标记为 sending防止重复取出 db_exec(UPDATE outbox SET status1 WHERE id?, row.id); // 3. 构造 MQTT 报文发送并等待应用层确认 result mqtt_publish_and_wait_ack(row.topic, row.payload, 5000); if (result ACK_OK) { // 4. 服务器确认成功直接删除记录 db_exec(DELETE FROM outbox WHERE id?, row.id); } else { // 5. 发送失败更新重试次数和下次扫描时间 db_exec( UPDATE outbox SET status0, retry_countretry_count1, next_retry_time? WHERE id?, now_ms() retry_backoff(retry_count), row.id); } } } }重试时间的计算我推荐指数退避加一点抖动。比如第一次重试等 5 秒第二次 10 秒第三次 20 秒上限 5 分钟。不要用固定间隔因为网络恢复往往有一个过程高频重试只会白白占用带宽和 CPU。抖动是为了防止多条记录在同一时刻同时重试造成流量尖峰。还有一个小细节status 标记为 1sending以后如果应用在等待 ACK 的 5 秒内崩溃了这条记录会一直停留在 sending 状态。所以在扫描条件里我加了 status IN (0,1)同时把 next_retry_time 在过去时间点的 sending 记录重新捞出来处理。宁可重复发送也不能让它卡死在 sending 里永远发不出去。这就是网络通信里的 at-least-once 语义。3.4 服务器侧要配合做幂等因为发送侧是 at-least-once 语义服务器在接收这些数据的时候必须做幂等处理。最直接的办法是网关每一条 outbox 记录生成一个全局唯一的消息 ID服务器收到数据后以这个消息 ID 作为主键或唯一索引去重。哪怕同一条数据因为网关崩溃恢复被重发了三遍服务器最终存储的数据也只有一份。具体到实现上可以在 outbox 表里加一个 msg_id 字段内容用 UUID 或时间戳加随机数生成。MQTT 消息的 payload 里包含这个 msg_id服务器解析 payload 后检查 Redis 或者数据库里有没有处理过。没有就正常入库有就回一个 ACK 但不重复写库。这一步如果省了生产环境大概率会碰到重复数据。不要觉得概率低网关断网重连那一刻最后一两条消息重发是非常常见的。3.5 积压太多怎么办限量发送和优先级处理断网时间一长outbox 表里积压的数据可能达到几十万条。这时如果网络恢复发送模块如果一次性把所有数据全捞出来网关内存会瞬间被打爆。所以扫描的时候必须限制每轮取出的条数我的经验是一轮不超过 200 条或者不超过 1MB 的 payload。发完这一批再扫下一批。这样即使有大量积压也只是发送耗时变长系统不会崩溃。优先级的问题也要考虑。有些现场数据是实时报警类有些是历史记录类。如果 mixed 在一个 outbox 表里断网恢复后按时间顺序补发历史数据报警信息会被延迟很久。我常用的做法是拆两个表或者在同一张表里加一个 priority 字段。报警类记录的 priority 设为 1扫描发送时先按 priority、再按 create_time 排序。这样既保证历史数据不丢又不耽误实时告警的及时性。4. 常见问题与排查技巧实录4.1 数据重复到达服务器库里出现了双份记录这是我在项目里被客户怼得最多的问题。第一次出现的时候我看日志发现 MQTT 客户端在重连成功那一刻之前发送超时但 broker 实际上已经处理掉的消息被网关重发了一遍。服务器没做幂等结果库里多了一批重复记录。排查思路分两层。先看网关侧是否真的产生了重复的 outbox 记录可以通过 msg_id 的 unique 约束来验证。再看服务器侧有没有做消息 ID 去重。我发现最有效的修法就是在服务器上给 msg_id 建唯一索引并采用先查后插或insert ignore策略。网关侧则要把发送确认机制落实到位只有服务器显式回 ACK 的记录才能删除TCP 发送成功不算数。4.2 断网恢复后历史数据补发顺序乱掉了数据乱序通常不是 outbox 表本身的问题而是发送线程在并发发送时没有控制好序号。如果你开了多个发送线程同时扫同一个 outbox 表不同线程取到的时间靠前的记录可能和后取到的记录交错发送服务器落库顺序自然就乱了。对大多数工控场景来说同一台设备的数据如果乱序曲线图就会出现毛刺和回跳。解决思路有两个。一是发送模块只开单线程严格按 create_time 顺序逐条发。二是如果对吞吐量有要求可以按设备 ID 做哈希分桶每个桶一个线程保证同一设备的记录严格有序。第二个方案对采集周期短、点位数量大的场景更实用。4.3 SQLite 在掉电瞬间丢记录的坑SQLite 在嵌入式设备上有一个经典坑如果建库时没有设置好同步模式掉电可能导致最近几条已提交事务丢失。默认的 synchronousFULL 时SQLite 在提交事务时要等数据真正落到磁盘才返回掉电安全最好但性能会慢一些而 synchronousNORMAL 在 WAL 模式下性能好很多但断电时可能丢失最后几个事务。工业网关掉电是常态我的建议是不要为了性能把 synchronous 降到 OFF尽量保留 FULL 或者至少 NORMAL 加 WAL。而且网关最好配一个小的 UPS 或者带掉电检测功能在检测到电压跌落时程序先停止新一轮采集把 outbox 表的数据紧急刷盘然后进入安全的关机流程。没有这个硬件配合任何软件方案都没法保证绝对的掉电零丢失。4.4 outbox 表太大磁盘被撑满怎么处理极端情况下如果网络连续断了几天网关本身存储空间又只有几百 MBoutbox 表是有可能撑满的。磁盘满了以后采集线程往数据库写入也会失败导致新数据都无法落库这会让问题雪上加霜。所以一定要给 outbox 表设置容量上限和淘汰策略。我用过一个比较实用的组合策略给 outbox 表按时间分区比如一天一个分区同时设置最大保留时间和最大行数。当超出阈值时先把超过 7 天的记录导出一个 CSV 备份文件再从表里删除。这样既不会丢历史数据也不会让新数据写不进去。读取侧发送模块要对表被清理的情况做防护如果记录没了就跳过继续发后面的别因为一条删除掉的记录卡死整个发送循环。4.5 云平台找不到网关最后几分钟的数据还有一种常见情况是网络是通的数据也在实时发送但是服务器侧不知道什么原因漏收了最后几分钟的数据。这种问题最隐蔽因为网关 outbox 表自己可能已经清空了——它收到了 ACK 就删除了记录。等到服务器那边发现数据有洞日志早就找不到了。排查我一般分三步。第一步确认服务器 ACK 报文的生成时间对比数据写入时间看是不是网络延迟导致 ACK 回得特别晚第二步检查网关的本地历史数据表里有没有这几分钟的数据如果网关侧落库了那就是发送或者服务器存储环节的问题第三步在服务器入库接口加 Trace ID 全链路日志把从 MQTT 收到报文到落库成功的每一步时间戳打出来定位到底哪一步丢了。最后提醒一下给 outbox 加上记录清理延迟策略也有帮助比如 ACK 成功后不要立刻删除记录而是先标记为 sent 状态保留 24 小时再异步清理。这样服务器如果哪天发现数据有洞还能倒查网关日志找原因。5. 我用这套方案重新改造网关后的一些体会改造完成之后我最直观的感受是再也不用盯着网络抖动看监控了。以前网络一抖我就紧张现在只要 outbox 表里没积压我基本不关心链路什么时候恢复。数据要么躺在本地的 outbox 表里要么已经发给服务器并被确认不会出现第三种我不知道它在哪的状态。这种确定感在工业现场比什么都值钱。最后再分享一个具体的小技巧。如果你用 SQLite 做 outbox 存储一定要定时执行 VACUUM 或者用自动 checkpoint 控制 WAL 文件大小。我遇到过几次 WAL 文件涨到好几个 GB导致磁盘告警的情况。最开始没注意后来在采集线程每次提交事务后加了一个 WAL 大小的检查超过阈值就主动执行一次 checkpoint问题就再也没出现过。这是个很不起眼的细节但对网关这种常年不断电的设备来说存储层的健康直接决定了 Outbox 模式能稳定跑多久。