应用层连接断了第一时间想的往往不是「为什么断」而是「赶紧重连」。automatic_reconnect就是 Paho 给客户端准备的方案断线后自动把连接建回来不用你写重试循环。但开了开关之后事情没那么简单。设备断线后多久会重连重试间隔固定还是越来越长重连成功后之前的订阅还收得到消息吗不假思索地打开automatic_reconnect大概率会在某个凌晨被「设备在线却收不到消息」的工单叫醒。这篇文章顺着自动重连的实现路径回答三个问题自动重连由什么状态控制为什么主动disconnect()之后它不会再连回来重试间隔怎么算指数退避和随机抖动在代码里长什么样重连成功后恢复了什么为什么「传输层连上了」不等于「会话层恢复了」沿着startConnectRetry状态机 → 重连命令入队 → 重连成功后的回调处理这条线逐段拆解 Paho 的实现。先给结论自动重连由shouldBeConnected用户意图和retrying重连中两个标志控制只有异常断开才触发主动disconnect()会清零shouldBeConnected确保不会重连。重试间隔是「指数退避 随机抖动」1s → 2s → 4s → … 翻倍到上限再在 ±20% 区间内抖动避免一群设备同时断线后同步重连造成惊群。重连只修复传输层连接会重新建立但订阅列表、离线消息是否还在取决于会话层clean sessiontrue时每次重连都是「一张白纸」。开启方式与重连状态机automatic_reconnect 默认关闭自动重连默认是关的。MQTTAsync_connectOptions的初始化宏里automaticReconnect 0如果你不显式打开断线后就停在断开状态什么都不会发生// paho.mqtt.c/src/MQTTAsync.h:1390#defineMQTTAsync_connectOptions_initializer{\{M,Q,T,C},8,60,1,65535,...,0,1,60...\}automaticReconnect字段在宏里默认是 0minRetryInterval是 1maxRetryInterval是 60。在 paho.mqtt.cpp 里用 builder 打开或显示指定退避区间autoconnOptsmqtt::connect_options_builder().clean_session().automatic_reconnect()// 默认 min1s, max60s.finalize();// 显示指定退避区间mqtt::connect_options_builder().automatic_reconnect(1s,60s)// minRetryInterval1s, maxRetryInterval60s重连状态机startConnectRetry自动重连的真正入口是MQTTAsync_startConnectRetry它在连接失败或断开时被调用负责计算下一次重连要等多久。// paho.mqtt.c/src/MQTTAsyncUtils.cvoidMQTTAsync_startConnectRetry(MQTTAsyncs*m){if(m-automaticReconnectm-shouldBeConnected){m-lastConnectionFailedTimeMQTTTime_start_clock();if(m-retrying)// 已在重连中指数退避1s → 2s → 4s → 8s → ... → 60s上限m-currentIntervalBasemin(m-currentIntervalBase*2,m-maxRetryInterval);else{// 首次重连从最小间隔开始m-currentIntervalBasem-minRetryInterval;m-retrying1;// 标记进入重连状态}// 加入随机抖动避免多客户端同时重连m-currentIntervalMQTTAsync_randomJitter(m-currentIntervalBase,m-minRetryInterval,m-maxRetryInterval);}}两个关键状态shouldBeConnected用户调connect()后置 1用户调disconnect()主动断开后置 0。它表达的是「用户希望保持连接」的意图。只有它仍为 1重连才会启动所以主动断开后不会重连。retrying首次失败从minRetryInterval开始指数级退避直到maxRetryInterval。注意currentInterval不是直接把currentIntervalBase拿去用而是再经过MQTTAsync_randomJitter抖动intMQTTAsync_randomJitter(intcurrentIntervalBase,intminInterval,intmaxInterval){constintmax_sleep(int)(min(maxInterval,currentIntervalBase)*1.2);constintmin_sleep(int)(max(minInterval,currentIntervalBase)/1.2);if(min_sleepmax_sleep)// shouldnt happen, but just in case{returnmin_sleep;}{intr;intrangemax_sleep-min_sleep1;constintbucketsRAND_MAX/range;constintlimitbuckets*range;do{rrand();}while(rlimit);{constintrandResultr/buckets;returnmin_sleeprandResult;}}}这段代码详细的说明在如何生成指定范围内的随机整数currentInterval通过MQTTAsync_randomJitter计算在[base/1.2, base×1.2]区间内随机抖动避免多个客户端同时断线后同步重连造成的惊群。最终等待时间在[base/1.2, base×1.2]内随机取值。为什么加抖动一批设备同时掉线时会用相同的退避节奏重连形成「同步重连」所有设备的负载同时压到 broker 上。随机抖动把这些峰值打散。一个值得注意的细节MQTTAsync_startConnectRetry这里只计算下一次重连的等待时间本身不发起重连。真正的「把连接请求塞回去」发生在重试循环里。重连命令如何入队MQTTAsync_startConnectRetry只算了「等多久」谁来发起真正的重连答案是在发送线程MQTTAsync_sendThread里的定时检查。sendThread 主循环发送线程的主循环逻辑很简单先处理命令队列里已有的命令然后等 1 秒再检查一次超时// paho.mqtt.c/src/MQTTAsyncUtils.c:sendThread 主循环节选while(!MQTTAsync_tostop){// 先处理队列中已有的命令while(command_count0){if(MQTTAsync_processCommand()0)break;// 没有命令可处理进入等待...command_countMQTTAsync_commands-count;}if((rcThread_wait_evt(send_evt,timeout))!0rc!ETIMEDOUT)...timeout1000;// 后续等待 1 秒MQTTAsync_checkTimeouts();// 每 1 秒检查一次}checkTimeouts真正的重连触发器MQTTAsync_checkTimeouts每 1 秒被 sendThread 调一次但函数内部还有一个 3 秒节流两次实际执行之间至少隔 3 秒。然后遍历所有客户端对处于重连状态automaticReconnect retrying的客户端做判断staticvoidMQTTAsync_checkTimeouts(void){if(MQTTTime_difftime(now,last)(DIFF_TIME_TYPE)3000)gotoexit;while(ListNextElement(MQTTAsync_handles,current)){//... 检查断连if(m-automaticReconnectm-retrying){if(m-reconnectNow||MQTTTime_elapsed(m-lastConnectionFailedTime)(ELAPSED_TIME_TYPE)(m-currentInterval*1000)){// 把 connect 命令插到队列头部MQTTAsync_queuedCommand*connmalloc(sizeof(MQTTAsync_queuedCommand));memset(conn,\0,sizeof(MQTTAsync_queuedCommand));conn-clientm;conn-commandm-connect;if(m-c-MQTTVersionMQTTVERSION_DEFAULT)conn-command.details.conn.MQTTVersion0;// 如果注册了 updateConnectOptions在这里回调可以刷新 token/密码if(m-updateConnectOptions){/**/}MQTTAsync_addCommand(conn,sizeof(m-connect));// 插入队列头m-reconnectNow0;}}}}这里有三个值得注意的点复用首次连接的完整命令。conn-command m-connect重连用的就是首次connect()时保存的完整连接命令包括 clean_session、keepAliveInterval、will 等参数。所以「重连时自动带上和首次一样的配置」是靠这个结构体保存实现的。版本 DEFAULT 时重置版本号。MQTTVERSION_DEFAULT(0)意味着「先试 3.1.1失败回退 3.1」。如果重连时不清零可能会沿用上次协商的版本这里显式重置为 0让每次重连都重新走一遍版本协商。updateConnectOptions回调用来刷新凭证。重连经常发生在长时间运行之后用户名/密码/token 可能已过期。MQTTAsync_checkTimeouts在添加重连命令前回调updateConnectOptions让用户有机会替换凭证回调里分配的新 username/password 会被替换进 m-c。如果你的 token 会过期这是处理凭证续期的入口。一旦命令插入队列头sendThread 下一轮循环就会把它取出来执行走正常的连接流程发送 CONNECT → 等 CONNACK。重连命令会插到队列头优先执行。但如果你在connection_lost回调里又手动调connect()就多塞了一条连接命令进去跟自动重连竞争执行。两个办法只能选一个要么用自动重连要么在 connection_lost 里手动连接。重连成功后发生了什么接收到服务端的 CONNACK 后receiveThread 调用MQTTAsync_completeConnection完成状态收尾// MQTTAsyncUtils.c精简staticintMQTTAsync_completeConnection(MQTTAsyncs*m,Connack*connack){if(m-c-connect_stateWAIT_FOR_CONNACK){if((rcconnack-rc)MQTTASYNC_SUCCESS){// 连接成功// 1. 清除重连标志下次失败从 minRetryInterval 重新开始m-retrying0;// 2. 更新连接状态m-c-connected1;m-c-good1;m-c-connect_stateNOT_IN_PROGRESS;// 3. 清理会话clean session 或 sessionPresent0if(m-c-cleansession||m-c-cleanstart)MQTTAsync_cleanSession(m-c);elseif(m-c-MQTTVersionMQTTVERSION_3_1_1connack-flags.bits.sessionPresent0)MQTTAsync_cleanSession(m-c);// 4. 重发断连期间堆积的消息if(m-c-outboundMsgs-count0){// 重置所有消息的 lastTouch// regardless1 表示无视重试间隔立即重试MQTTProtocol_retry(zero,1,1);}//... MQTTv5 服务端可能覆盖 keepalive// 唤醒 sendThreadThread_signal_evt(send_evt);}}}这里有两个值得注意的细节retrying 0重连成功后退出「重连中」状态currentIntervalBase的翻倍计数也就此归零下次断线会从minRetryInterval重新开始退避。本地会话状态清理不仅是 broker 端有 clean session 语义客户端本地也会清理cleansession/cleanstart为true时直接清即使没设只要 v3.1.1 且 CONNACK 的标志位sessionPresent 0该字段表示 broker 端没有旧会话客户端也把自己的本地状态清掉避免残留。随后触发用户回调。注意connected回调的cause参数会告诉你这次是首次连接还是自动重连// MQTTAsyncUtils.c:2148-2199receiveThread 处理 CONNACK 的后续rcMQTTAsync_completeConnection(m,connack);if(rcMQTTASYNC_SUCCESS){// 1.调用 onSuccess / onSuccess5 回调if(m-connect.onSuccess){data.alt.connect.serverURI...;data.alt.connect.MQTTVersion...;data.alt.connect.sessionPresentsessionPresent;(*(m-connect.onSuccess))(m-connect.context,data);m-connect.onSuccessNULL;// 清空防止重复调用m-connect.onFailureNULL;}elseif(m-connect.onSuccess5){// ... v5 类似但包含 properties 和 reasonCode}// 2. 调用 connected 回调if(m-connected){char*reasononSuccess?connect onSuccess called:automatic reconnect;(*(m-connected))(m-connected_context,reason);// ^^^^^^^// 自动重连时 reason automatic reconnect// 首次连接时 reason connect onSuccess called}// 3. MQTT v5: 应用服务端返回的 RECEIVE_MAXIMUM更新最大处理消息数if(m-c-MQTTVersionMQTTVERSION_5){if(MQTTProperties_hasProperty(connack-properties,MQTTPROPERTY_CODE_RECEIVE_MAXIMUM)){intrecv_max(int)MQTTProperties_getNumericValue(connack-properties,MQTTPROPERTY_CODE_RECEIVE_MAXIMUM);if(m-c-maxInflightMessagesrecv_max)m-c-maxInflightMessagesrecv_max;}}}on_success和connected的区别前者携带连接结果详情URI、版本、sessionPresent用于判断「broker 端还有没有我的旧会话」后者是「连接已就绪」的事件钩子cause 为 “automatic reconnect” 时说明这是自动重连完成的通常在这里做订阅恢复等业务初始化。自动重连解决的是传输层TCP 链路重建、MQTT 握手完成。但业务上下文在会话层断线期间 broker 是否还替你保留会话取决于 clean session / clean start 怎么设。这就是长连接系统最常见的问题重连成功后broker 端要是已经没有你的会话sessionPresent false订阅列表就是空的message_arrived不会被触发连接看着正常实际消息一个都收不到。自动重连 clean sessiontrue每次重连 broker 都给你一个全新会话订阅和离线消息全部归零。想保留离线消息就不能用 clean session。正确的做法通常是在connected()回调里重新订阅或者先检查on_success里的sessionPresent再决定要不要恢复订阅。总结自动重连的完整链路断线keepalive 超时 / TCP 错误 │ └──→ startConnectRetryshouldBeConnected 为 1 才启动 │ 退避 currentIntervalBase 翻倍 抖动 currentInterval │ └──→ sendThread 每 1s 调 checkTimeouts内部 3s 节流 │ 到点 → 把 m-connect 命令插入队列头可先回调 updateConnectOptions 刷新凭证 │ └──→ 重连成功completeConnection 置 retrying0 → onSuccesssessionPresent/ connectedcauseautomatic reconnect → v5 应用 RECEIVE_MAXIMUM → 订阅是否恢复看 clean session 与会话是否过期回看开头问题的结论重连由shouldBeConnectedretrying两个标志控制主动disconnect()不会触发重连重试间隔是指数退避 随机抖动±20%防惊群重连只修复传输层订阅/离线消息是否还在由会话层决定clean sessiontrue时每次重连都是新的会话。