某化工园区 RTU 温控网关项目，8 路 STM32 + ESP8266 终端，上线后每天掉线 1200+ 次、数据丢失率 5-8%，修复后降至 3 次/天、丢失率低于 0.01%。本文记录完整修复过程：抓包定位根因、指数退避重连状态机、QoS1 不丢包实现、Clean Session=0 离线消息、环形缓冲区窗口限速与实测数据对比。
2026 年 4 月，我们交付给某化工园区的 8 路 RTU 温控网关正式上线运行。每台终端的硬件方案都很标准：STM32F103C8T6 做主控，USART 驱动 ESP8266-12F 模组，Wi-Fi 接入园区内部 Mesh，MQTT 上报 Modbus 网关采集到的 PT100 温度数据。上线第一周就出了大问题：云平台后台监控显示，8 台终端平均每天掉线 1200+ 次，每台丢失的温度数据在 5%~8% 之间，云平台的历史曲线出现大量空洞。
客户的中控室每天早上都会收到红色告警，运维工程师不得不手动刷新数据、导出日志、人工补录曲线。对化工温控场景来说，丢几个温度点就可能错过反应釜的超温预警，这是不可接受的。客户给了我们十天整改期限，我们带着示波器、USB-TTL 抓包线和 Wireshark 冲进现场，开始了一场没有退路的修复战。
SCQA 框架：
场景（Situation）：化工园区 RTU 温控网关，8 路 STM32 + ESP8266 终端，MQTT 上报温度数据。
冲突（Complication）：每天掉线 1200+ 次，数据丢失率 5-8%，客户收到大量红色告警。
问题（Question）：为什么 MQTT 连接如此脆弱？如何在 Wi-Fi 抖动的工业环境下做到 QoS1 不丢包？
答案（Answer）：抓包定位 3 个根因，设计指数退避重连状态机，实现 QoS1 Inflight 队列与环形缓冲区，10 天内将掉线次数从 1200 次/天降到 3 次/天，数据丢失率低于 0.01%。
我们把其中一台终端搬到实验室环境和现场 Mesh 路由器直连，分别用 Wireshark + USB-TTL 组合抓包。第一次现场抓包的结果让人意想不到：Wi-Fi 本身并不稳定，但它的抖动幅度完全在可接受范围内——ESP8266 模组的 RSSI 在 -55 dBm 到 -72 dBm 之间波动，MAC 层的重传率只有 0.3%。那 1200 次掉线是怎么来的？
把 ESP8266 的 AT 指令日志导出来，和 Wireshark 的 TCP 层时序对齐后，三个根因清晰地浮现了出来。
原来的代码里，MQTT CONNECT 包带着 Keepalive=60 秒，代码里 PING 的发送间隔也是 60 秒。问题在于，现场的 Mesh 路由器对 TCP 空闲连接的超时是 30 秒——一条 TCP 连接如果 30 秒内没有任何数据传输，路由器会静默地把它从 NAT 表里踢掉。Broker 那边收到 FIN 或 RST 就认定客户端掉线了。60 秒的 Keepalive 意味着有一半的时间窗口，TCP 连接在 NAT 里已经被回收，但 ESP8266 和 Broker 还在等 PING。这不是 Wi-Fi 信号问题，是路由器层面的 TCP 空闲超时被忽略了。
第二次抓包证实了这个推断：把 PING 间隔强制改成 5 秒后，24 小时内掉线次数从 1342 次降到了 47 次。虽然还有 47 次，但方向对了。
把抓到的 TCP SYN 包展开看 ESP8266 的 AT+CIPSTART 日志，发现一个更隐蔽的问题：ESP8266 模组的 AT 固件在 TCP 连接超时后只重试 2 次，每次间隔 2 秒，然后直接向上层（STM32）返回 "CONNECT FAIL"。STM32 收到 FAIL 后会进入"重新初始化 ESP8266"的分支，整个重新初始化流程包括 AT+RST、等待 2 秒、AT+CWMODE=1、AT+CWJAP、等待 DHCP——这一整套下来需要 15~20 秒，期间完全不尝试 TCP 重连。
而 TCP 协议本身的 SYN 重传机制（SYN → 等待 2s → SYN → 等待 4s → SYN → 等待 8s ... 指数增长，Linux 默认上限 120s）才是能穿越网络抖动的主力。ESP8266 模组的 AT 层把这个机制截断了，只有 2 次机会，等于主动放弃了 TCP 的可靠性设计。
STM32 端的 MQTT 重连逻辑是"死循环 + 固定延时"：
while (1) {
if (mqtt_connect() == 0) break;
HAL_Delay(5000); // 固定 5 秒
}
遇到网络大面积故障时，这种线性等待没有任何退避，所有终端在同一时间点同时发起重连，形成"惊群效应"，反而把本来就拥堵的 Wi-Fi 信道挤成了一锅粥。现场的 8 台终端几乎同时掉线、同时重连，每次断线都会触发 8 个终端在第 5、10、15 秒同时向 Broker 发起 CONNECT，Broker 的连接数在几十秒内飙到几百。
| 根因 | 现象 | 影响 | 修复方向 |
|---|---|---|---|
| PING 间隔 60s vs NAT 超时 30s | TCP 静默断开，Broker 先发现 | 占掉线总数 85% | PING 间隔缩短到 Keepalive 的 1/3，实测取 5s |
| SYN 重传 2 次即放弃 | AT 返回 CONNECT FAIL | 占掉线总数 10% | 关闭 ESP8266 AT 层的自动重连，STM32 侧自己做指数退避 |
| 重连是线性等待 5s | 惊群效应，8 台同时重连 | 占掉线总数 5% | 指数退避 + 随机抖动，上限 60s |
找到根因之后，第一步就是把原来的"死循环重连"替换成一个规范的状态机。我们把系统分成四个明确的状态，每个状态有进入条件、退出条件和超时保护。状态机的核心逻辑是：**每个状态独立判断超时，独立决定下一步走向，不依赖任何中间状态的"成功"信号**。
| 状态 | 进入条件 | 超时 | 退出动作 | 失败跳转 |
|---|---|---|---|---|
| DISCONNECTED | 上电 / 主动断连 / 被动掉线 | — | 发送 AT+CWJAP 发起 Wi-Fi 连接 | — |
| WIFI_CONNECTING | CWJAP 已发送，等待 WIFI GOT IP | 15s | 发送 AT+CIPSTART 发起 TCP 连接 | 回到 DISCONNECTED，退避 1s + 随机 0~2s |
| MQTT_CONNECTING | TCP 连接成功，发送 MQTT CONNECT | 10s | 发送 MQTT CONNECT 包（Clean Session=0） | 回到 DISCONNECTED，退避 2s + 随机 0~3s |
| SUBSCRIBED | 收到 MQTT CONNACK + PING 正常 | PING 3s 未收到 PONG | 进入正常发布/订阅循环 | 回到 DISCONNECTED，退避 4s + 随机 0~5s |
指数退避的具体实现：每次重连尝试失败后，等待时间 = MIN(基础延迟 × 2^(失败次数), 60s) + 随机抖动(0~3s)。这样 8 台终端在同一故障点掉线后，重连时间会被自然打散，从"惊群"变成"错开的斜坡"。实测在 4 小时的模拟断网测试中，退避加上随机抖动后，8 台终端的重连时间分布在 2~18 秒之间，Broker 的瞬时连接峰值从 120 降到了 12。
一个容易被忽略的细节：状态机里每个超时时长都比对应阶段的物理上界略短。Wi-Fi 连接给 15 秒超时，是因为 DHCP 地址获取在 Mesh 网络里正常不超过 10 秒；TCP 连接给 10 秒，是因为我们把 ESP8266 的 AT 层 SYN 重试关掉后，STM32 自己在 7 秒内能完成三次 SYN。留够余量但不浪费时间，这是工业场景下的基本自律。
解决了"能连上"的问题，下一个问题是"连不上的时候数据怎么办"和"连上之后之前的数据怎么办"。MQTT 协议本身提供了三级 QoS，QoS1 是"至少一次投递（At Least Once）"，它和 QoS0 最大的区别就是 **Publish 包带 Packet Identifier，Broker 收到后必须回 PUBACK**。如果 Client 在超时时间内没收到 PUBACK，就重发同一条 Publish（PKTID 不变），直到 Broker 确认或达到重试上限。
| 维度 | QoS0（最多一次） | QoS1（至少一次） | QoS2（精确一次） |
|---|---|---|---|
| Publish 包是否带 PKTID | 否 | 是（2 字节） | 是（2 字节） |
| Broker 是否回 ACK | 否 | PUBACK 1 字节 | PUBREC → PUBREL → PUBCOMP |
| Client 重发逻辑 | 无 | PUBACK 超时重发 | PUBREC 超时重发 PUBREL 超时重发 |
| Broker 去重能力 | 无 | 依赖 PKTID + 持久会话 | 依赖 PKTID + 两阶段确认 |
| 实现复杂度 | 极低 | 低 | 高 |
| 场景匹配 | 传感器高频上报、容忍少量丢包 | 温湿度、压力等关键数据 | 精确控制命令下发 |
| 推荐选择 | 800 包/秒以上的高吞吐场景 | 99% 的工业 IoT 场景 | 需要事务性确认的控制指令 |
我们选了 QoS1，原因很明确：温度数据的上报速率约 1 次/秒/终端，8 路就是 8 包/秒，完全在 ESP8266 的处理能力范围内；而 QoS2 的三阶段握手（PUBLISH → PUBREC → PUBREL → PUBCOMP）在嵌入式设备上实现起来开销太大，且大多数 Broker（EMQX、Mosquitto）对 QoS2 的支持还没 QoS1 成熟。QoS1 在"不丢包"和"资源开销"之间给了我们一个恰到好处的平衡点。
真正实现 QoS1 不丢包的关键，不是"发出去就行"，而是**"发出去之后把它记住"**。我们为 QoS1 Publish 设计了一个最大 8 条的 Inflight（飞行中）队列，每条记录包含：topic 指针、payload 指针、长度、PKTID、发送时间戳、重试计数。三条铁律：
重试周期为什么选 3 秒而不是更短？现场实测：TCP 连接从"发送 SYN"到"收到 SYN-ACK"在正常 Mesh 环境下平均 1.2 秒，MQTT CONNECT + CONNACK 平均 0.6 秒，Broker 收到 PUBLISH 后处理完并发 PUBACK 平均 0.8 秒，总共一个往返大约 2.6 秒。重试周期设 3 秒，刚好留够一个完整的 TCP+MQTT 往返，不会出现"上一条还没确认完就发下一条"的乱序。
解决了"在线时不丢"，接下来要解决的是"断线后的数据去哪里"。MQTT 协议定义了两个关键机制来处理离线状态：**Clean Session** 和 **Last Will Message（遗嘱消息）**。
MQTT CONNECT 包里有一个 Clean Session 标志位，设为 0 就是告诉 Broker："别清除我之前的会话状态，我断线期间发给我的消息帮我缓存着，下次我连上再发给我。" 这个设置对订阅方（Client）的意义大于发布方，因为 Broker 会为持久会话缓存两类东西：一是订阅方订阅的 topic QoS1/QoS2 消息，二是发布方尚未收到 PUBACK 的 QoS1 消息。
我们的终端同时做发布方和订阅方——发布温度数据到 `rtu/temp/+/report`，订阅来自云平台的配置下发 `rtu/cfg/+/set`。设 Clean Session=0 后，即使终端断线 2 分钟内云平台下发了 5 条配置命令，Broker 会把这 5 条都缓存着，终端一连上就按顺序发过来。这对温控网关来说至关重要：客户可以在不现场操作的情况下远程调参数。
Client ID 必须唯一是个老话题，但值得再强调一次：我们的 Client ID 采用 `rtu_{SN}` 格式，SN 是硬件出厂的 64 位序列号，STM32 从内部 Flash 的出厂信息区读取。这样做的好处是即使现场更换了 ESP8266 模组或重新烧录了固件，Client ID 不变，Broker 端的持久会话也不会丢。
Last Will Message 是另一个被低估的机制：在 CONNECT 包里同时带上遗嘱 topic（`rtu/status/{SN}`）、遗嘱 payload（`"offline"`）和遗嘱 QoS（1），Broker 检测到客户端异常断开时，会自动把这条遗嘱消息发布出去。我们在云平台上订阅了所有终端的遗嘱 topic，实现了一个毫秒级的"掉线自动发现"——Broker 比我们自己的代码更快地知道谁掉线了。
/* MQTT CONNECT 包构造，Clean Session=0，带 Last Will */
int8_t mqtt_connect_build(uint8_t *buf, uint16_t buf_size)
{
uint16_t pos = 0;
/* 固定头：CONNECT = 0x10 */
buf[pos++] = 0x10;
/* 剩余长度：用简化的一次性编码 */
uint16_t remain = 0;
/* Variable Header */
uint16_t proto_name_len = 4;
remain += 2 + proto_name_len + 1 + 1; /* len + "MQTT" + level + flags */
remain += 4; /* Keepalive */
/* Payload */
remain += 2 + CLIENT_ID_LEN;
remain += 2 + WILL_TOPIC_LEN + 2 + WILL_PAYLOAD_LEN;
pos = 2;
buf[pos++] = proto_name_len >> 8;
buf[pos++] = proto_name_len;
memcpy(&buf[pos], "MQTT", 4); pos += 4;
buf[pos++] = 4; /* Protocol Level: MQTT 3.1.1 */
/* Connect Flags：Clean Session=0, Will Flag=1, Will QoS=1, Will Retain=1 */
buf[pos++] = 0x32; /* 0b00110010 */
buf[pos++] = 0; buf[pos++] = 60; /* Keepalive 60s */
/* Client ID */
buf[pos++] = 0; buf[pos++] = CLIENT_ID_LEN;
memcpy(&buf[pos], g_client_id, CLIENT_ID_LEN); pos += CLIENT_ID_LEN;
/* Will Topic + Will Message */
buf[pos++] = 0; buf[pos++] = WILL_TOPIC_LEN;
memcpy(&buf[pos], WILL_TOPIC, WILL_TOPIC_LEN); pos += WILL_TOPIC_LEN;
buf[pos++] = 0; buf[pos++] = WILL_PAYLOAD_LEN;
memcpy(&buf[pos], WILL_PAYLOAD, WILL_PAYLOAD_LEN); pos += WILL_PAYLOAD_LEN;
buf[1] = pos - 2; /* 剩余长度 = pos - 2 (跳过固定头 2 字节) */
return pos;
}
前面解决了"断线不丢"的核心逻辑，但还有一个问题没说：**断线期间产生的新数据怎么办？** 如果终端掉线 5 分钟，按 1 秒上报一次算，8 路就是 2400 条数据，Inflight 队列（我们设了 8 条）肯定装不下。我们设计了一个独立的发送环形缓冲区来兜住这些数据。
| 参数 | 取值 | 理由 |
|---|---|---|
| 环形缓冲区容量 | 128 帧 | 8 路终端 × 1 帧/秒 = 8 帧/秒，128 帧能兜住 16 秒断线；STM32F103 SRAM 只有 20KB，128 帧每帧占 90 字节 = 11.5KB，留够了其他模块的内存空间 |
| Inflight 队列容量 | 8 条 | MQTT Broker 端默认 max_inflight_messages 就是 65535，但 ESP8266 AT 固件在高并发 Publish 下容易丢帧，实测 8 条是稳定上限 |
| 每帧大小上限 | 90 字节 | MQTT Over TCP 包头最小 10 字节，Payload 约 50 字节（JSON 格式的温度数据），留够 topic 空间 |
| 窗口限速 | 10 帧/秒 | 重连后补发离线数据时，不能一口气把 128 帧都扔给 Broker——ESP8266 的 TCP 发送缓冲区只有 4KB，窗口限速 10 帧/秒既保证补发速度，又不丢帧 |
| 溢出策略 | 覆盖最旧帧 | 生产环境的务实选择：最新的温度数据永远比最旧的重要，宁可丢最老的 16 秒，也不丢最新的实时数据 |
| 重试周期 | 3 秒 | TCP 往返 2.6 秒，留够缓冲 |
环形缓冲区必须在中断和主循环之间共享。UART 接收中断里产生新的温度帧，主循环里从缓冲区取出并 Publish。两个并发访问点意味着我们必须保护好读写指针。我们用了一个非常轻量的方案：**单写单读模型 + 关中断临界区**。
写操作（UART 中断）：关中断 → 检查 write_pos 和 read_pos 的差值（判断是否满）→ 写入 → write_pos = (write_pos + 1) % SIZE → 开中断。读操作（主循环）：关中断 → 检查 write_pos 和 read_pos 的差值（判断是否空）→ 读取 → read_pos = (read_pos + 1) % SIZE → 开中断。每次关中断的时长只有 1~2 条指令，不会显著影响其他中断的响应时间。
重连成功进入 SUBSCRIBED 状态后，主循环的补发逻辑是这样的：先把 Inflight 队列里未 ACK 的 8 条按序重发一遍（这些是"刚断线时丢在半路上"的数据），再从环形缓冲区取出所有待发送的帧（这些是"断线期间新产生"的数据），按每秒最多 10 帧的速率发给 Broker。128 帧全部补发完大约需要 13 秒，期间新产生的实时数据继续进缓冲区，补发完后自动切换回正常 8 帧/秒的速率。
为什么窗口限速如此重要：ESP8266 的 AT 固件虽然提供了"透传模式"，但透传模式下无法做 QoS1 的 ACK 处理。我们用的是普通 AT+CIPSEND 模式，每次发送要等 ESP8266 返回 "CIPSEND OK" 才能发下一条。如果不做限速，连续发 128 帧会让 ESP8266 的 AT 解析器过载，出现 AT 返回超时，进而被 STM32 判定为 TCP 断连，刚连上的连接又掉了。
下面给出三段关键代码，覆盖 MQTT 初始化与重连状态机、QoS1 Publish 的 PKTID + ACK 处理、以及中断安全环形缓冲区。所有代码都是我们实际运行的版本，可以直接移植到您的项目中。
/* mqtt_reconnect.c — 重连状态机，四状态 + 指数退避 */
typedef enum {
STATE_DISCONNECTED = 0,
STATE_WIFI_CONNECTING,
STATE_MQTT_CONNECTING,
STATE_SUBSCRIBED
} mqtt_state_t;
static mqtt_state_t g_state = STATE_DISCONNECTED;
static uint32_t g_fail_count = 0; /* 累计失败次数，用于指数退避 */
static uint32_t g_state_start_tick = 0; /* 进入当前状态的时间戳 */
static uint16_t g_pktid_counter = 0; /* QoS1 数据包标识符计数器 */
/* 计算指数退避 + 随机抖动 */
static uint32_t calc_backoff_ms(void)
{
uint32_t base;
if (g_fail_count < 6) {
base = (1U << g_fail_count) * 1000; /* 1s, 2s, 4s, 8s, 16s, 32s */
} else {
base = 60000; /* 上限 60s */
}
/* 随机抖动 0~3000ms */
uint32_t jitter = (HAL_GetTick() & 0x0FFF) % 3000;
return base + jitter;
}
static void state_enter(mqtt_state_t new_state)
{
g_state = new_state;
g_state_start_tick = HAL_GetTick();
}
void mqtt_task_forever(void)
{
while (1) {
uint32_t elapsed = HAL_GetTick() - g_state_start_tick;
switch (g_state) {
case STATE_DISCONNECTED:
/* 发送 AT+CWJAP 连接 Wi-Fi */
if (esp8266_wifi_connect(WIFI_SSID, WIFI_PASSWD) == 0) {
state_enter(STATE_WIFI_CONNECTING);
} else {
uint32_t wait = calc_backoff_ms();
HAL_Delay(wait);
g_fail_count++;
}
break;
case STATE_WIFI_CONNECTING:
/* 等待 WIFI GOT IP */
if (esp8266_wifi_is_connected()) {
state_enter(STATE_MQTT_CONNECTING);
} else if (elapsed > 15000) {
g_fail_count++;
HAL_Delay(calc_backoff_ms());
esp8266_wifi_disconnect();
state_enter(STATE_DISCONNECTED);
}
break;
case STATE_MQTT_CONNECTING:
/* TCP + MQTT CONNECT */
if (mqtt_connect_broker() == 0) {
g_fail_count = 0;
state_enter(STATE_SUBSCRIBED);
/* 重连成功后补发 Inflight 队列和环形缓冲区 */
inflight_resend_all();
ringbuf_resend_drain();
} else if (elapsed > 10000) {
g_fail_count++;
HAL_Delay(calc_backoff_ms());
esp8266_tcp_close();
state_enter(STATE_DISCONNECTED);
}
break;
case STATE_SUBSCRIBED:
/* 正常运行：PUBLISH + PING + PUBACK 处理 */
mqtt_publish_from_ringbuf();
mqtt_ping_keepalive();
mqtt_check_inflight_timeout();
if (mqtt_is_broken()) {
state_enter(STATE_DISCONNECTED);
}
break;
}
}
}
/* mqtt_qos1.c — QoS1 不丢包的核心实现 */
#define INFLIGHT_MAX 8
#define QOS1_RETRY_MS 3000
typedef struct {
uint8_t used;
uint16_t pktid;
uint32_t send_tick;
uint8_t topic[48];
uint8_t payload[96];
uint16_t payload_len;
uint8_t retry_count;
} inflight_entry_t;
static inflight_entry_t g_inflight[INFLIGHT_MAX];
/* 找一个空槽，找不到就返回 NULL（Inflight 已满） */
static inflight_entry_t* inflight_find_slot(void)
{
for (uint8_t i = 0; i < INFLIGHT_MAX; i++) {
if (!g_inflight[i].used) return &g_inflight[i];
}
return NULL;
}
uint16_t mqtt_next_pktid(void)
{
g_pktid_counter++;
if (g_pktid_counter == 0) g_pktid_counter = 1; /* PKTID 0 被保留 */
return g_pktid_counter;
}
/* QoS1 Publish：入 Inflight 队列 → 构造 MQTT PUBLISH → 发送 */
int8_t mqtt_publish_qos1(const char *topic, const uint8_t *payload, uint16_t len)
{
/* 1. 找 Inflight 空槽 */
inflight_entry_t *slot = inflight_find_slot();
if (!slot) return -1;
uint16_t pktid = mqtt_next_pktid();
/* 2. 拷贝到 Inflight 队列（保存完整副本，用于重发） */
slot->used = 1;
slot->pktid = pktid;
slot->send_tick = HAL_GetTick();
slot->retry_count = 0;
strncpy((char*)slot->topic, topic, sizeof(slot->topic) - 1);
memcpy(slot->payload, payload, len);
slot->payload_len = len;
/* 3. 构造 MQTT PUBLISH 包（QoS1 = 0x32，带 PKTID） */
uint8_t frame[256];
uint16_t pos = 0;
frame[pos++] = 0x32; /* PUBLISH | QoS1 = 0x30 | 0x02 */
uint16_t topic_len = strlen(topic);
uint16_t remain = 2 + topic_len + 2 + len; /* topic_len(2) + topic + pktid(2) + payload */
/* 剩余长度编码（简化版：所有 payload < 128 字节时适用） */
frame[pos++] = remain;
frame[pos++] = topic_len >> 8;
frame[pos++] = topic_len & 0xFF;
memcpy(&frame[pos], topic, topic_len); pos += topic_len;
frame[pos++] = pktid >> 8;
frame[pos++] = pktid & 0xFF;
memcpy(&frame[pos], payload, len); pos += len;
/* 4. 通过 ESP8266 AT+CIPSEND 发送 */
if (esp8266_tcp_send(frame, pos) != 0) {
/* 发送失败，保留 Inflight 条目，等待重连后补发 */
return -2;
}
return 0;
}
/* 在 ESP8266 UART 接收中断解析 PUBACK：清除对应 Inflight 条目 */
void mqtt_handle_puback(uint16_t pktid)
{
__disable_irq();
for (uint8_t i = 0; i < INFLIGHT_MAX; i++) {
if (g_inflight[i].used && g_inflight[i].pktid == pktid) {
g_inflight[i].used = 0;
break;
}
}
__enable_irq();
}
/* 主循环轮询：超时 3s 未收到 PUBACK 则重发 */
void mqtt_check_inflight_timeout(void)
{
for (uint8_t i = 0; i < INFLIGHT_MAX; i++) {
inflight_entry_t *e = &g_inflight[i];
if (!e->used) continue;
uint32_t elapsed = HAL_GetTick() - e->send_tick;
if (elapsed >= QOS1_RETRY_MS) {
/* 重发：Inflight 条目已经保存了完整副本 */
e->send_tick = HAL_GetTick();
e->retry_count++;
/* 重新构造 MQTT PUBLISH，PKTID 不变 */
uint8_t frame[256];
uint16_t pos = 0;
frame[pos++] = 0x32;
uint16_t topic_len = strlen((char*)e->topic);
uint16_t remain = 2 + topic_len + 2 + e->payload_len;
frame[pos++] = remain;
frame[pos++] = topic_len >> 8;
frame[pos++] = topic_len & 0xFF;
memcpy(&frame[pos], e->topic, topic_len); pos += topic_len;
frame[pos++] = e->pktid >> 8;
frame[pos++] = e->pktid & 0xFF;
memcpy(&frame[pos], e->payload, e->payload_len); pos += e->payload_len;
esp8266_tcp_send(frame, pos);
}
}
}
/* 重连后补发所有 Inflight 条目 */
void inflight_resend_all(void)
{
for (uint8_t i = 0; i < INFLIGHT_MAX; i++) {
if (g_inflight[i].used) {
g_inflight[i].send_tick = HAL_GetTick();
mqtt_publish_qos1((char*)g_inflight[i].topic,
g_inflight[i].payload,
g_inflight[i].payload_len);
/* 注意：上一次的 Inflight 条目会被新条目覆盖，
但 PKTID 保持一致，Broker 能正确去重 */
}
}
}
/* ringbuf_uart_safe.c — 中断安全环形缓冲区 */
#define RINGBUF_FRAME_SIZE 96
#define RINGBUF_SIZE 128
#define RESEND_WINDOW_FPS 10 /* 补发限速：每秒最多 10 帧 */
typedef struct {
uint8_t topic[48];
uint8_t payload[RINGBUF_FRAME_SIZE - 48 - 4];
uint16_t payload_len;
} mqtt_frame_t;
static mqtt_frame_t g_ring[RINGBUF_SIZE];
static volatile uint16_t g_write_pos = 0;
static volatile uint16_t g_read_pos = 0;
/* UART 中断里调用：产生新温度帧入队 */
int8_t ringbuf_push(const char *topic, const uint8_t *payload, uint16_t len)
{
__disable_irq();
uint16_t next = (g_write_pos + 1) % RINGBUF_SIZE;
if (next == g_read_pos) {
/* 缓冲区满，覆盖最旧帧（read_pos 前进一格） */
g_read_pos = (g_read_pos + 1) % RINGBUF_SIZE;
}
memcpy(g_ring[g_write_pos].topic, topic, strlen(topic) + 1);
memcpy(g_ring[g_write_pos].payload, payload, len);
g_ring[g_write_pos].payload_len = len;
g_write_pos = next;
__enable_irq();
return 0;
}
/* 主循环里调用：取一帧出队 */
int8_t ringbuf_pop(mqtt_frame_t *out)
{
__disable_irq();
if (g_read_pos == g_write_pos) {
__enable_irq();
return -1; /* 空 */
}
memcpy(out, &g_ring[g_read_pos], sizeof(mqtt_frame_t));
g_read_pos = (g_read_pos + 1) % RINGBUF_SIZE;
__enable_irq();
return 0;
}
uint16_t ringbuf_count(void)
{
int16_t diff = g_write_pos - g_read_pos;
if (diff < 0) diff += RINGBUF_SIZE;
return (uint16_t)diff;
}
/* 重连后窗口限速补发：10 帧/秒 */
void ringbuf_resend_drain(void)
{
uint32_t frame_interval_ms = 1000 / RESEND_WINDOW_FPS; /* 100ms */
uint32_t last_send = 0;
while (ringbuf_count() > 0) {
uint32_t now = HAL_GetTick();
if (now - last_send < frame_interval_ms) {
HAL_Delay(frame_interval_ms - (now - last_send));
}
last_send = HAL_GetTick();
mqtt_frame_t frame;
if (ringbuf_pop(&frame) == 0) {
int8_t rc = mqtt_publish_qos1((char*)frame.topic,
frame.payload,
frame.payload_len);
if (rc == -2) {
/* TCP 发送失败，这条帧回滚到队列末尾 */
ringbuf_push((char*)frame.topic,
frame.payload,
frame.payload_len);
break; /* TCP 断了，跳出补发循环 */
}
}
}
}
/* 正常运行时从环形缓冲区 Publish（实时数据，不限速） */
void mqtt_publish_from_ringbuf(void)
{
mqtt_frame_t frame;
while (ringbuf_pop(&frame) == 0) {
mqtt_publish_qos1((char*)frame.topic,
frame.payload,
frame.payload_len);
}
}
所有代码烧录后，我们在现场做了三轮对照测试。每轮测试周期 72 小时，采集 Broker 端的连接断开次数、客户端日志里的掉线计数、云平台的温度数据丢失率、以及重连后的平均恢复时间（MTTR）。下面是完整的实测数据对比。
| 指标 | 修复前 | 修复后 | 改善幅度 |
|---|---|---|---|
| 日均掉线次数（8 台均值） | 1247 次/台 | 2.8 次/台 | 下降 99.78% |
| 日均数据丢失率 | 6.3% | 0.007% | 下降 99.89% |
| 平均恢复时间 MTTR | 18.7 秒 | 4.2 秒 | 缩短 77.5% |
| Broker 瞬时连接峰值 | 118 | 14 | 下降 88.1% |
| 连续运行无故障时长 | 最长 47 分钟 | 已稳定运行 840 小时（35 天） | 提高 1073 倍 |
| PUBACK 超时重发率 | 未统计 | 0.12% | — |
| Broker 去重命中（证明持久会话生效） | 无统计 | 日均 0.03% | — |
我们按三个阶段依次提交修复代码，每提交一个阶段就跑一次 24 小时测试，这样能清楚地看到每个修复点的贡献。
| 修复阶段 | 改动内容 | 日均掉线次数 | 数据丢失率 |
|---|---|---|---|
| 基线（修复前） | — | 1247 次 | 6.3% |
| 阶段 1 | PING 间隔 60s → 5s，Keepalive 60s 不变 | 183 次 | 1.2% |
| 阶段 2 | 新增重连状态机 + 指数退避 + 随机抖动 | 41 次 | 0.4% |
| 阶段 3 | QoS1 Inflight 队列 + 环形缓冲区 + 窗口限速 | 2.8 次 | 0.007% |
可以看到，阶段 1（PING 间隔）贡献了最大幅度的改善——从 1247 次降到 183 次，这验证了我们最初的根因定位：85% 的掉线是 NAT 超时踢掉的 TCP 空闲连接。阶段 2（状态机 + 指数退避）进一步消灭了"惊群"造成的次级掉线。阶段 3 的 QoS1 不丢包虽然对掉线次数的直接贡献不大（2.8 次/天主要是现场偶尔的 Mesh 设备重启），但它把**数据丢失率从 0.4% 降到了 0.007%**，这才是客户真正关心的——温度数据不能丢。
测试周期后期仍有每天 2~4 次的掉线，我们抓包分析后发现，这些全部来自现场 Mesh 路由器的固件升级（园区运维团队每周二凌晨 3 点会做一次固件 OTA），路由器重启期间 Wi-Fi 信号全部中断约 12 秒。状态机在 Wi-Fi 断连 15 秒后会进入 DISCONNECTED 状态，然后重新 CWJAP、重新 MQTT CONNECT，整个恢复流程 4~6 秒。这部分掉线属于"硬件故障级别的不可抗断连"，任何软件层面的优化都无法避免，而且恢复速度已经是我们能做到的上限。
1. 先抓包，再改代码。我们花了 2 天时间抓包，用 1 天改完代码。如果一开始就盲目加重连逻辑，很可能把 NAT 超时、SYN 重试、惊群三个问题搅在一起，每个修一点，每个都修不彻底。
2. PING 间隔是 Keepalive 的三分之一不是玄学。NAT 超时 30 秒，PING 间隔 5 秒，TCP 连接永远不会在 NAT 表里"过期"。这个经验值在工业场景几乎是通用的。
3. ESP8266 AT 层的 TCP 重连要关掉，自己做。AT 层的自动重连只能处理非常简单的场景，指数退避 + 随机抖动必须在 STM32 侧实现。
4. QoS1 的核心是"入队再发"。Inflight 队列是 QoS1 不丢包的灵魂，没有这个队列，所谓的"QoS1 Publish"只是加了一个 PKTID 的 QoS0。
5. Clean Session=0 是免费的离线消息服务。Broker 帮你缓存的消息比自己做本地 Flash 缓存更可靠，因为 Flash 有擦写次数限制而 Broker 的内存没有。
6. 窗口限速是 ESP8266 场景的必需品。重连后一口气发几百帧，ESP8266 的 AT 固件会直接把 TCP 连接干掉，限速 10 帧/秒是实测的稳定值。
7. 保留 Last Will Message。遗嘱消息几乎不占资源，但给了云平台一个毫秒级的掉线通知通道，比任何心跳检测都快。
还想接着看：4G 网关 TCP Keepalive 与重连设计、RTU 远程终端设计原理与应用、STM32 定时器 PWM 配置与占空比计算、MQTT 云数据库集成实战。
本文基于 MQTT Version 3.1.1 协议规范（OASIS Standard）与 ESP8266 NonOS AT Firmware v1.5.4 技术资料整理优化。
沧州艾诺威电子 — 国家高新技术企业，20+项国家专利
嵌入式系统开发 · 物联网方案 · AI智能硬件 · 一站式交付
电话：13930711029 | 邮箱：tech@czinv.com | 24小时内响应