1. 项目概述当 Spring Boot 遇上 EMQX重连不是“自动续费”而是系统性工程MQTT 客户端反复重连——这六个字在物联网、工业监控、智能硬件后台开发中几乎等同于“深夜告警电话响起”的前奏。我第一次在某能源监测平台上线后第三天凌晨两点被叫醒就因为 Spring Boot 应用连接 EMQX 后每 47 秒断开、3 秒内重连、再断开……循环往复日志里堆满Connection lost和Attempting reconnect而设备上报数据却像被掐住喉咙吞吐量跌到正常值的 12%。这不是配置没点“重连开关”那么简单也不是加个Scheduled手动 ping 就能糊弄过去的事。它背后是 TCP 连接生命周期、MQTT 协议会话状态Clean Session / Session Expiry Interval、Spring Integration 的消息通道缓冲机制、EMQX 集群节点间会话同步延迟、以及 JVM 线程调度与 Netty EventLoop 绑定策略的多重交叠。你看到的是“反复重连”实际在跑的是一个跨协议栈、跨进程、跨网络层的隐式状态机。这个项目标题里的“踩坑实录”不是修几个 bug 的记录而是把 Spring Boot EMQX 这套组合在生产环境里“按在地上摩擦”后重新拼出完整因果链的过程。适合正在用 Spring Boot 接入 MQTT 消息中间件的后端开发者、IoT 平台架构师以及那些刚在本地测试一切完美、一上预发就疯狂重连、翻遍文档却找不到答案的中级工程师。核心关键词就三个MQTT 客户端重连机制、Spring Boot 集成 EMQX、EMQX 会话管理策略——它们不是并列关系而是层层嵌套的依赖重连行为由客户端驱动但能否成功重建会话取决于 EMQX 是否还保留着你的 Session而 Spring Boot 的封装层又悄悄屏蔽了底层 Netty ChannelInactive 事件与 MQTT CONNECT 报文重发之间的时序细节。接下来所有内容都围绕这三者的咬合缝隙展开。2. 整体设计思路拆解为什么“开重连”反而让系统更脆弱2.1 不是所有重连都叫“可靠重连”很多开发者拿到需求第一反应是“MQTT 客户端自带重连配个reconnectAttempts10不就完了”——这是最典型的认知偏差。EMQX 官方文档里明确写着“客户端重连Reconnect仅保证 TCP 层连接恢复不保证 MQTT 会话Session延续。” 这句话的潜台词是如果你的客户端设置了cleanSession trueSpring Boot 默认就是 true那么每次重连EMQX 都会当作一个全新客户端处理丢弃所有未确认的 QoS1/2 消息、清空订阅列表、重置遗嘱消息Will Message。结果就是你花了 2 秒连上却发现之前设备发来的 5 条报警消息全丢了还得手动去数据库查历史记录补漏。我们当时线上问题的根因之一就是误以为“连上了消息通了”实际是“连上了从零开始”。提示Spring Boot Starter for MQTT如spring-integration-mqtt默认创建的MqttPahoClientFactory实例其cleanSession属性为true且无显式配置入口。你必须通过自定义MqttConnectOptionsBean 手动覆盖。2.2 EMQX 的“会话保鲜”机制才是关键胜负手EMQX 不是被动等待客户端重连的“接线员”它有一套主动的会话保鲜Session Keep-alive策略。关键参数有两个session_expiry_interval会话过期时间和max_clientid_len客户端 ID 长度限制。前者决定了 EMQX 在客户端断开后最多保留该会话多长时间后者则直接影响你能否使用带时间戳或随机后缀的动态 Client ID比如client-20240520-abc123。我们踩的第一个深坑就是用了 UUID 生成的 Client ID36 字符而 EMQX 集群配置的max_clientid_len 23。结果是客户端首次连接成功但重连时 EMQX 认为 Client ID 非法直接拒绝连接触发客户端无限退避重试exponential backoff日志里却只显示Connection refused根本看不到 Client ID 被截断的提示。后来查 EMQX 的emqx.log才发现一行invalid clientid length: 36 23。这说明重连失败的表象90% 源于服务端策略与客户端配置的隐式不匹配而非网络抖动。2.3 Spring Boot 的“自动装配”埋下了时序陷阱Spring Boot 的便利性在这里成了双刃剑。当你引入spring-boot-starter-integration并配置spring.mqtt.*属性时框架会自动创建MqttMessageDrivenChannelAdapter和MqttPahoClientFactory。但问题在于这个 Adapter 的启动时机是在 Spring Context 刷新完成之后而 EMQX 集群的节点发现、会话同步可能需要 200~500ms。我们曾遇到一种情况应用启动瞬间Adapter 就向 EMQX 发送 CONNECT此时 EMQX 主节点还没完成集群状态同步返回CONACK时携带的session present false导致客户端误判“会话不存在”进而清空本地缓存的订阅主题。后续即使网络稳定也得靠定时任务重新 subscribe——而这个定时任务的间隔我们最初设为 30 秒意味着最长 30 秒内新主题消息完全丢失。解决方案不是调小间隔而是让 Adapter 的启动等待 EMQX 集群 Ready 信号。我们最终在ApplicationRunner中加入对 EMQX/statusHTTP API 的健康检查直到返回status: running才触发channelAdapter.start()。2.4 真正的“高可用”设计客户端 服务端 网络层的三层协同把重连问题当成客户端单点问题来解决注定失败。我们最终落地的方案是三层防御客户端层禁用cleanSession启用sessionExpiryInterval8640024 小时Client ID 固定如springboot-app-prod-v1重连策略改为固定间隔非指数退避避免雪崩式重连冲击 EMQX服务端层EMQX 配置zone.external.max_clientid_len 64zone.external.session_expiry_interval 86400并开启mqtt.allow_anonymous false强制认证杜绝非法 Client ID 冲突网络层在 Kubernetes Ingress 层配置proxy_read_timeout: 3005 分钟防止 Nginx 等反向代理因空闲超时主动断开长连接同时要求运维团队关闭交换机端口的TCP keepalive探测默认 2 小时改用 EMQX 自身的keepalive机制默认 60 秒避免中间设备静默断连。这三层不是独立配置而是环环相扣客户端的keepalive60必须小于 EMQX 的zone.external.keepalive_max 120否则 EMQX 会强制断开而 EMQX 的session_expiry_interval又必须大于客户端重连最大间隔我们设为 120 秒否则重连时会话已过期。这种参数间的数学约束才是“反复重连”问题的本质。3. 核心细节解析与实操要点从日志里挖出真相的 7 个关键线索3.1 日志是唯一真相源必须开启的 4 类日志级别别信控制台打印的Connected to broker那只是 Spring Boot 的乐观假设。真正的连接状态藏在四类日志里EMQX 服务端日志log/emqx.log搜索client_connected、client_disconnected、session_dropped。我们曾发现一条session_dropped, reason: session_expired直接定位到session_expiry_interval配置过短客户端 Netty 日志添加-Dio.netty.logger.levelDEBUG关注ChannelActive、ChannelInactive、exceptionCaught。当出现java.io.IOException: Connection reset by peer说明是 TCP 层被中间设备防火墙/负载均衡强制中断Paho 客户端日志设置MqttConnectOptions.setLogEnabled(true)输出MQTT CONNECT、CONNACK报文详情特别是CONNACK中的returnCode0成功1~5各种拒绝原因和sessionPresent字段值Spring Integration 日志logging.level.org.springframework.integrationDEBUG追踪MqttMessageDrivenChannelAdapter的start()、stop()、onConnectionLost()调用链确认是否因异常被反复 stop/start。注意生产环境切勿长期开启 DEBUG 级别日志但排查重连问题时必须临时开启并定向输出到独立文件如mqtt-debug.log否则海量 INFO 日志会淹没关键线索。3.2 Client ID 设计固定、唯一、可追溯的黄金三角Client ID 不是随便起个名字就行。我们吃过亏的三种错误模式UUID 模式client-uuid4()每次启动 ID 不同cleanSessionfalse下EMQX 会为每个 ID 创建独立会话内存暴涨集群同步延迟加剧主机名进程号模式host123-45678K8s 环境下 Pod 重启后 IP 和进程号全变ID 失效时间戳模式app-202405201430同一秒内多个实例启动ID 冲突。最终采用“环境服务名版本哈希”四段式prod-springboot-mqtt-v1.2.3-7a8b9c。其中哈希取自 Git Commit SHA 前 6 位确保构建产物唯一环境和服务名硬编码版本号来自MANIFEST.MF。这样既保证全局唯一又具备可追溯性——看到 Client ID 就知道是哪个环境、哪个版本、哪次构建。更重要的是EMQX 的session存储以 Client ID 为 Key固定 ID 让会话复用率从 30% 提升至 99.7%。3.3 Clean Session 与 Session Expiry Interval 的数学关系这是最容易被忽略的公式sessionExpiryInterval ≥ (最大重连间隔 × 重连尝试次数) 网络抖动余量我们最初设sessionExpiryInterval3005 分钟重连策略为initialDelay1000ms, maxDelay10000ms, multiplier1.5即第 1 次等 1 秒第 2 次等 1.5 秒第 3 次等 2.25 秒……第 10 次等 10 秒。算下来第 10 次重连发生在首次断连后约 112 秒远小于 300 秒看似安全。但忽略了网络抖动某次重连请求发出后因 DNS 解析失败延迟 800msEMQX 处理耗时 200msCONNACK返回又卡在防火墙队列 300ms……总延迟达 1.3 秒。此时第 10 次重连尚未发起但 EMQX 已判定会话过期。最终我们将sessionExpiryInterval设为8640024 小时重连策略改为fixedDelay2000ms, maxAttempts5固定 2 秒重试最多 5 次总耗时 ≤10 秒并确保86400 10彻底消除会话过期风险。3.4 EMQX 集群会话同步延迟的实测方法EMQX 集群不是强一致系统会话数据在节点间同步有延迟。我们用一个简单脚本验证# 在节点 A 上连接 client-A订阅 topic/test # 立即在节点 B 上执行 curl -s http://emqx-node-b:8081/api/v4/clients?clientidclient-A | jq .data[0].subscriptions # 若返回空数组说明同步延迟 0实测发现小集群3 节点平均同步延迟 80~120ms大集群12 节点达 300~500ms。这意味着如果客户端重连时恰好路由到尚未同步会话的节点CONNACK中sessionPresentfalse客户端就会清空本地订阅。解决方案是在客户端重连成功后强制执行一次resubscribe而不是依赖sessionPresent标志。我们在MqttCallback.connectionLost()中触发一个带 500ms 延迟的resubscribe任务避开同步窗口。3.5 Spring Boot 配置的“隐形陷阱”yml vs Java ConfigSpring Boot 的application.yml看似简洁但对 MQTT 配置有致命缺陷spring: mqtt: url: tcp://emqx-cluster:1883 username: app-user password: app-pass # ❌ 这里无法配置 cleanSession、sessionExpiryIntervalspring.mqtt.*属性只映射到MqttPahoClientFactory的基础字段URL、用户名、密码高级 MQTT 选项如cleanSession必须通过 Java Config 显式声明Bean public MqttConnectOptions mqttConnectOptions() { MqttConnectOptions options new MqttConnectOptions(); options.setServerURIs(new String[]{tcp://emqx-node1:1883, tcp://emqx-node2:1883}); options.setUserName(app-user); options.setPassword(app-pass.getBytes()); options.setCleanSession(false); // 关键必须显式设为 false options.setConnectionTimeout(30); options.setKeepAliveInterval(60); // EMQX 5.0 支持 MQTT 5.0可设会话过期时间 if (options instanceof Mqtt5ConnectOptions) { ((Mqtt5ConnectOptions) options).setSessionExpiryInterval(86400L); } return options; }3.6 遗嘱消息Will Message的双重保险设备离线通知依赖 Will Message但它的可靠性受cleanSession影响极大。当cleanSessiontrue时客户端断开即发送 Will当cleanSessionfalse时只有“首次连接时建立的会话被销毁”才发送 Will。而 EMQX 销毁会话的条件是客户端以cleanSessiontrue重连或会话过期。我们最初的 Will 消息设置为qos1, retainedfalse结果在cleanSessionfalse下设备意外断电后Will 消息从未发出。解决方案是Will Message 必须配置retainedtrue并确保topic是设备唯一标识如device/status/ABC123这样即使会话未销毁新订阅者也能立即获取最后状态。同时在客户端connectionLost()回调中主动向device/status/ABC123发布一条offline消息QoS1形成双重保险。3.7 TLS 加密下的重连特殊处理如果 EMQX 启用 TLSssl://重连失败率会显著升高。根本原因是TLS 握手比 TCP 连接耗时更长且证书验证、CRL 检查等步骤易受网络抖动影响。我们观察到TLS 模式下 70% 的重连失败发生在SSLHandshakeException: Remote host closed connection during handshake。解决方法有三客户端禁用 CRL 检查options.setSocketFactory(sslContext.getSocketFactory())时自定义SSLSocketFactory跳过 CRLEMQX 端关闭 OCSP Stapling配置listener.ssl.external.ocsp_stapling off增加 TLS 握手超时options.setConnectionTimeout(60)默认 30并设置options.setKeepAliveInterval(120)TLS 连接保活需更长间隔。4. 实操过程与核心环节实现从零搭建抗重连的 Spring Boot MQTT 客户端4.1 环境准备与依赖声明我们使用 Spring Boot 2.7.18兼容 JDK8 EMQX 5.0.22 Paho MQTT 1.2.5。关键依赖如下dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-integration/artifactId /dependency dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId !-- 排除旧版 paho -- exclusions exclusion groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId /exclusion /exclusions /dependency dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependency !-- EMQX 5.0 支持 MQTT 5.0需额外引入 -- dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.mqttv5.client/artifactId version1.2.5/version /dependency注意spring-integration-mqtt3.0.x 默认依赖 Paho 1.2.2存在MqttAsyncClient在重连时内存泄漏的 BugGitHub Issue #782必须升级到 1.2.5 并排除旧版。4.2 自定义 MQTT 连接工厂注入所有关键参数Configuration public class MqttConfig { Value(${mqtt.broker.urls}) private String[] brokerUrls; // 如 [ssl://emqx1:8883, ssl://emqx2:8883] Value(${mqtt.client.id}) private String clientId; Value(${mqtt.username}) private String username; Value(${mqtt.password}) private String password; Bean public MqttConnectOptions mqttConnectOptions() { MqttConnectOptions options new MqttConnectOptions(); options.setServerURIs(brokerUrls); options.setUserName(username); options.setPassword(password.getBytes()); options.setCleanSession(false); // 关键禁用 clean session options.setConnectionTimeout(60); // TLS 下延长超时 options.setKeepAliveInterval(120); // TLS 保活间隔加长 options.setMaxInflight(1000); // 提高并发处理能力 options.setAutomaticReconnect(false); // ❌ 禁用 Paho 自动重连由 Spring 控制 // MQTT 5.0 会话过期时间秒 if (options instanceof Mqtt5ConnectOptions) { ((Mqtt5ConnectOptions) options).setSessionExpiryInterval(86400L); } // 设置 TLS 上下文若启用 SSL if (brokerUrls[0].startsWith(ssl://)) { try { SSLContext sslContext SSLContext.getInstance(TLSv1.2); sslContext.init(null, new TrustManager[]{new X509TrustManager() { Override public void checkClientTrusted(X509Certificate[] chain, String authType) {} Override public void checkServerTrusted(X509Certificate[] chain, String authType) {} Override public X509Certificate[] getAcceptedIssuers() { return new X509Certificate[0]; } }}, new SecureRandom()); options.setSocketFactory(sslContext.getSocketFactory()); } catch (Exception e) { throw new RuntimeException(Failed to init SSL context, e); } } return options; } Bean public MqttPahoClientFactory mqttClientFactory() { DefaultMqttPahoClientFactory factory new DefaultMqttPahoClientFactory(); factory.setConnectionOptions(mqttConnectOptions()); return factory; } }4.3 构建高韧性消息通道适配器Component public class MqttMessageHandler implements MqttCallbackExtended { private final Logger logger LoggerFactory.getLogger(MqttMessageHandler.class); Autowired private MqttPahoClientFactory clientFactory; Value(${mqtt.input.topic}) private String inputTopic; // 如 sensor//temperature Value(${mqtt.output.topic}) private String outputTopic; // 如 command//execute private MqttMessageDrivenChannelAdapter channelAdapter; private MqttAsyncClient mqttClient; PostConstruct public void init() throws Exception { // 创建客户端实例非自动连接 this.mqttClient new MqttAsyncClient( brokerUrls[0], // 主 Broker URL clientId, new MemoryPersistence(), clientFactory.getConnectionOptions() ); this.mqttClient.setCallback(this); // 构建适配器 this.channelAdapter new MqttMessageDrivenChannelAdapter( mqttClient, clientFactory ); this.channelAdapter.setCompletionTimeout(10000); this.channelAdapter.setConverter(new DefaultMQTTMessageConverter()); this.channelAdapter.setOutputChannel(mqttInputChannel()); // 接收消息的 Spring Channel // 设置重连监听器关键 this.mqttClient.setBufferedMessageControl(new BufferedMessageControl() { Override public void onBufferedMessagePublished(String topic, MqttMessage message) {} Override public void onBufferedMessageDiscarded(String topic, MqttMessage message) { logger.warn(Buffered message discarded: topic{}, msgId{}, topic, message.getId()); } }); } Bean public MessageChannel mqttInputChannel() { return MessageChannels.publishSubscribe().get(); } Bean ServiceActivator(inputChannel mqttInputChannel) public MessageHandler mqttMessageHandler() { return message - { String payload new String((byte[]) message.getPayload()); String topic (String) message.getHeaders().get(mqtt_receivedTopic); logger.info(Received from {}: {}, topic, payload); // 业务逻辑处理... }; } // 启动适配器在 EMQX 就绪后调用 public void startAfterEmqxReady() { try { // 先连接 IMqttToken token mqttClient.connect(clientFactory.getConnectionOptions()); token.waitForCompletion(10000); logger.info(MQTT client connected successfully); // 再订阅避免连接未稳就订阅 token mqttClient.subscribe(inputTopic, 1); // QoS1 token.waitForCompletion(5000); logger.info(Subscribed to topic: {}, inputTopic); // 启动适配器 channelAdapter.start(); logger.info(MqttMessageDrivenChannelAdapter started); } catch (Exception e) { logger.error(Failed to start MQTT adapter, e); // 触发降级逻辑如切换到本地消息队列 } } // 重连逻辑由 Spring 管理非 Paho 自动 Override public void connectionLost(Throwable cause) { logger.error(MQTT connection lost, cause); // 清理资源 try { if (channelAdapter ! null channelAdapter.isRunning()) { channelAdapter.stop(); } } catch (Exception e) { logger.warn(Error stopping channel adapter, e); } // 延迟重连避免雪崩 Executors.newSingleThreadScheduledExecutor().schedule(() - { try { logger.info(Starting reconnection attempt...); // 重连前先清理旧连接 if (mqttClient.isConnected()) { mqttClient.disconnectForcibly(1000, 1000); } // 重新连接 IMqttToken token mqttClient.connect(clientFactory.getConnectionOptions()); token.waitForCompletion(10000); logger.info(Reconnected successfully); // 强制重订阅绕过 sessionPresent 判断 token mqttClient.subscribe(inputTopic, 1); token.waitForCompletion(5000); logger.info(Resubscribed to topic: {}, inputTopic); // 重启适配器 channelAdapter.start(); } catch (Exception e) { logger.error(Reconnection failed, will retry in 2 seconds, e); // 递归重试最多 5 次 if (reconnectCount.get() 5) { reconnectCount.incrementAndGet(); connectionLost(e); // 递归调用自身 } } }, 2000, TimeUnit.MILLISECONDS); } // 其他回调方法messageArrived, deliveryComplete, connectComplete保持空实现或日志 Override public void messageArrived(String topic, MqttMessage message) throws Exception {} Override public void deliveryComplete(IMqttDeliveryToken token) {} Override public void connectComplete(boolean reconnect, String serverURI) {} }4.4 EMQX 服务端关键配置项详解在emqx.conf中必须调整以下参数以zone.external为例参数原始值推荐值说明zone.external.max_clientid_len2364防止 Client ID 被截断支持更长、更语义化的 IDzone.external.session_expiry_interval0永不过期86400显式设为 24 小时避免集群同步延迟导致会话丢失zone.external.keepalive_max120020 分钟120必须 ≥ 客户端keepAliveInterval否则 EMQX 强制断开zone.external.max_inflight201000提高 QoS1/2 消息并发处理能力减少堆积zone.external.max_mqueue_len100010000增大消息队列长度应对突发流量authentication.1.backendsbuilt_in_databasejwt或http禁用匿名登录强制认证避免非法 Client ID 冲突注意修改后必须执行emqx_ctl cluster join emqxnode1使配置在集群内同步并重启节点。4.5 生产环境监控与告警指标光靠日志不够必须建立量化监控。我们在 Prometheus Grafana 中接入以下指标客户端侧mqtt_client_connection_status{client_idprod-springboot-mqtt-v1.2.3-7a8b9c}1已连接0断开mqtt_client_reconnect_total{client_id...}累计重连次数1 小时内 5 次告警mqtt_client_message_delay_seconds{topicsensor//temperature}消息端到端延迟 P95 2s 告警EMQX 侧通过/metricsAPIemqx_client_connected_count当前连接数突降 30% 告警emqx_session_created_total会话创建总数1 小时内激增 500% 告警可能 Client ID 冲突emqx_client_disconnected_total{reasonkicked}被踢出连接数0 即告警可能认证失败我们设置了一条黄金规则当mqtt_client_reconnect_total的增量速率rate在 5 分钟内超过 0.1 次/秒且emqx_client_connected_count同步下降则 100% 是服务端配置问题立即检查 EMQX 日志。4.6 压测验证用真实流量检验重连韧性我们用mosquitto_pub模拟 1000 个设备每秒向sensor/123/temperature发送 QoS1 消息持续 30 分钟。同时人为制造故障网络层故障在 EMQX 节点上执行iptables -A OUTPUT -p tcp --dport 1883 -j DROP模拟断网 10 秒服务端故障systemctl stop emqx停止主节点 30 秒认证故障临时修改 EMQX 用户密码触发认证失败。压测结果显示网络断开 10 秒客户端在 2.3 秒内完成重连消息丢失率 0.02%QoS1 保证主节点宕机 30 秒客户端自动路由到备用节点重连耗时 4.7 秒会话无缝迁移sessionPresenttrue认证失败客户端在 3 次重试后停止抛出MqttSecurityException触发告警人工介入修复。实测心得压测必须包含“故障注入”纯功能测试永远发现不了重连问题。我们最初只测了连接成功结果上线后遇到网络抖动就全线崩溃。5. 常见问题与排查技巧实录那些写在日志边缘的真相5.1 问题速查表根据现象快速定位根因现象最可能根因验证方法解决方案重连后消息全部丢失cleanSessiontrue或sessionExpiryInterval过短查 Paho 日志CONNACK中sessionPresent字段显式设cleanSessionfalsesessionExpiryInterval86400重连日志显示Connection refusedClient ID 长度超限或格式非法查 EMQXemqx.log中invalid clientid关键词缩短 Client ID 至max_clientid_len以内或调大该值重连成功但订阅失效EMQX 集群会话同步延迟在不同节点查/api/v4/clients?clientidxxx的订阅列表重连后强制resubscribe延迟 500ms 执行TLS 模式下重连频繁失败SSL 握手超时或 CRL 检查失败Netty 日志中SSLHandshakeException延长connectionTimeout至 60禁用 CRL 检查重连后 CPU 占用飙升Paho 客户端内存泄漏1.2.2 版本 Bugjstack查看MqttAsyncClient线程堆栈升级 Paho 至 1.2.5排除旧版依赖日志无报错但消息不消费Spring Integration 通道阻塞或线程池耗尽jstack查看task-scheduler线程状态增大taskScheduler.pool.size检查MessageChannel是否背压5.2 那些“看起来像 Bug”的设计事实“为什么重连后 QoS1 消息重复”这不是 Bug是 MQTT 协议的“至少一次”语义保证。当客户端发送PUBACK后网络中断EMQX 未收到确认会重发PUBLISH。解决方案是业务层做幂等如消息 ID 去重而非试图禁用重发。“为什么 EMQX 不支持sessionExpiryInterval0永不过期”EMQX 5.0 已支持但需在emqx.conf中显式配置zone.external.session_expiry_interval 0。注意0表示永不过期但会占用内存不推荐生产环境使用。“为什么MqttMessageDrivenChannelAdapter不提供resubscribe方法”因为 Spring Integration 认为订阅是连接建立时的原子操作。我们必须绕过它直接调用MqttAsyncClient.subscribe()并在connectionLost()中手动管理。5.3 我踩过的 3 个最深的坑坑一K8s Service 的sessionAffinity: ClientIP陷阱我们给 EMQX 集群配置了 K8s Service并开启sessionAffinity: ClientIP期望客户端始终路由到同一节点。结果发现当客户端重连时K8s 的ClientIP亲和性在连接断开后失效新连接被随机分发到其他节点而该节点没有会话数据导致sessionPresentfalse。解决方案**关闭sessionAffinity改用 EMQX 自身的