ActiveMQ MQTT 连接风暴排查手记业务背景映美云打印平台在线云打印机 4 万 台涵盖云针式打印机、热敏打印机、喷墨电票打印机等多个产品线。打印机设备通过 MQTT 协议接入 ActiveMQ与上层应用通信。应用端主要包括映美云打印 App— 移动端打印入口下发打印订单e 标签— 标签打印业务产送标签模板与打印任务云驱动— PC 端云打印驱动模拟本地打印机体验映管家— 设备管理与运维平台管理打印机状态与配置消息流向各应用端生产/订阅打印状态和打印订单状态通过 ActiveMQ 的 Topic 模式推送到对应的打印机设备。设备上线时订阅自身的print/{设备号}主题应用端通过该主题下发打印任务设备回传打印状态。故障概述项目说明服务器4 核 8G中间件Apache ActiveMQ 6.0.1协议MQTT (mqttnio)在线设备4 万 台云打印机现象连接数飙升 3-4 万线程数爆炸到 2 万CPU 跑满 350%结果进程反复崩溃重启打印任务大面积延迟或失败排查周期约一周的持续排查与多轮优化一、问题初现Timer already cancelled最初的告警很模糊——客户端报 “Could not accept connection: Timer already cancelled”。监控脚本显示 CPU 不高当时监控脚本有 bugCPU 读数始终为 0但netstat一查吓一跳[rootxxx ~]# netstat -antp | grep java | grep 1883 | wc -l 348193 万 连接而当时配置的maximumConnections是默认值。第一反应是连接数超限了但深入看监控日志发现更诡异的现象线程数一度冲到 3088连接数忽高忽低像潮水一样涨落进程每隔几十分钟就崩一次监控脚本自动拉起来第一轮猜测错的一开始怀疑是 Linux 系统参数问题tcp_max_tw_buckets只有 5000somaxconn不够ulimit -n限制等等。调了一轮系统参数效果不大——问题根本不在系统层在应用层。二、定位根因ios-print 重复 clientId翻 activemq.log满屏的Stealing link for clientId ios-print。Stealing link for clientId ios-print from tcp://59.50.11.69:63080 Stealing link for clientId ios-print from tcp://59.50.11.69:63081 Stealing link for clientId ios-print from tcp://36.109.251.214:23456 ...每秒 7 次。所有 iOS 打印机设备都在用同一个 clientId “ios-print”连接 MQTT。MQTT 协议的坑MQTT 协议规定 clientId 必须唯一。当第二个连接用相同 clientId 连上来时ActiveMQ 默认行为是“Stealing link”—— 踢掉旧连接新连接顶上。正常情况下这没什么但如果有几千台设备共用一个 clientId就会发生设备A连接 → 注册 ios-print 设备B连接 → 踢掉设备A → 设备A断线重连 → 踢掉设备B 设备C连接 → 踢掉设备A → 设备B重连 → 踢掉设备C ...几千台设备互相踢形成连接风暴。每次踢连接都会创建新线程、销毁旧线程叠加closeAsyncfalse同步关闭和maxInactivityDuration10s超时断连太快线程数越积越多最终拖垮进程。为什么会有相同 clientId客户端代码写死了clientId ios-print每台 iOS 设备都用这个值连。正确做法是每台设备生成唯一 clientId比如用设备序列号但客户端发版周期长服务端必须先扛住。三、第一次尝试allowLinkStealing“false”失败最直觉的想法既然 Stealing link 是问题那禁止 Stealing 不就行了ActiveMQ 有个参数叫allowLinkStealingfalse放在transportConnector上。结果启动失败。原因是这个参数是 transport 层的 URI 参数不是 XML 属性而且 ActiveMQ 6.x 对这个参数的支持方式变了。直接加在 XML 标签上会解析报错。教训ActiveMQ 的 transport 参数要放在 URI 里不是 XML 属性里。四、第二次尝试自定义插件 RejectDuplicateClientIdPlugin既然配置层面做不到就写插件。思路很简单写一个 BrokerPlugin拦截addConnection()如果 clientId 在黑名单里如 ios-print且已有活跃连接直接拒绝拒绝时同步关闭底层 transport减少资源泄漏限频日志不要每次拒绝都打日志插件核心逻辑publicvoidaddConnection(ConnectionContextcontext,ConnectionInfoinfo)throwsException{StringclientIdinfo.getClientId();if(clientId!nullblockedClientIds.contains(clientId)){if(!activeBlockedIds.add(clientId)){// 已存在 → 拒绝logRateLimited(clientId);// 限频日志stopTransportSync(context);// 同步关 I/OthrownewIllegalStateException(clientId clientId already connected);}}super.addConnection(context,info);}插件设计的几个细节为什么不用InvalidClientIDException因为这个异常类在 activemq-broker 包里不一定有用标准的IllegalStateException更稳妥。为什么要stopTransportSync()同步关抛异常后 ActiveMQ 的 catch 块还会调一次stop()如果 transport 没关就走完整的同步清理路径很慢。先关了 I/O后面的 stop() 就是 no-op。为什么限频日志每秒 13 次拦截每次 3 行日志WARN 堆栈一天能写满磁盘。改成每 30 秒输出一条汇总[RejectDuplicateClientIdPlugin] 拒绝 ios-print 重复连接 (最近30秒内 735 次)配套log4j2 日志抑制插件自己的日志好控制但 ActiveMQ 内部在连接被拒绝时还会打一堆 WARNFailed to add Connection、Stopping、Transport Connection failed需要在 log4j2.properties 里把相关 logger 调到 ERRORlogger.transport.name org.apache.activemq.broker.TransportConnection logger.transport.level ERROR logger.transportConnector.name org.apache.activemq.broker.TransportConnector logger.transportConnector.level ERROR logger.mqttConverter.name org.apache.activemq.transport.mqtt.MQTTProtocolConverter logger.mqttConverter.level ERROR第三条是后来加的——批量断连时每个死连接会产生 32 行堆栈Broken pipe15 秒能产生 3000 行日志。五、最大的坑URI 多行书写导致参数全部失效插件部署了配置也改了但线上日志显示maximumConnections10000、closeAsyncfalse——都是默认值配置文件里写的参数一个都没生效。折腾了很久才发现原因activemq.xml 中 URI 跨多行书写换行和缩进空格被原样包含在 URI 字符串中。!-- ❌ 错误写法URI 跨多行空格被包含进属性值 --transportConnectornamemqttniourimqttnio://0.0.0.0:1883? maximumConnections20000amp;wireFormat.maxInactivityDuration10000amp;.../ActiveMQ 拿到的实际 URI 是mqttnio://0.0.0.0:1883?\n maximumConnections20000\n wireFormat.maxInactivityDuration10000\n ...参数名前面带了一堆空格日志里显示为%20%20%20%20%20URI 解析器匹配不上全部回退默认值。正确写法URI 必须在一行内。!-- ✅ 正确写法URI 一行写完 --transportConnectornamemqttniourimqttnio://0.0.0.0:1883?maximumConnections33000amp;transport.useInactivityMonitortrueamp;wireFormat.maxInactivityDuration60000amp;transport.closeIdleConnectionTimeout60000amp;transport.keepAliveTime15000amp;transport.soKeepAlivetrueamp;transport.tcpNoDelaytrueamp;transport.closeAsynctrueamp;allowLinkStealingfalse/这个坑浪费了好几天。每次改了配置重启以为生效了实际全是默认值在跑。验证参数是否生效的唯一方法看Listening for connections日志对比参数值。六、更深层的问题NIO 线程池无限制插件生效、closeAsynctrue、allowLinkStealingfalse都配置正确后连接数稳定在 3 万线程数正常 130 左右。但每次重启后 8-10 分钟必然出现一次线程爆炸15:48:47 THR105 CONN30138 CPU26% ← 正常 15:50:28 THR1822 CONN30564 CPU61% ← 突然跳升 16:02:24 THR15925 CONN30937 CPU236% ← 线程爆炸 16:06:16 THR21442 CONN32645 CPU340% ← 濒临崩溃327 个新连接产生了 1685 个新线程5 线程/连接严重不正常。原因ActiveMQ 的 NIO Selector 线程池没有上限。正常情况下 8 个线程处理 3 万连接没问题但重连风暴来临时每分钟 1 万 新连接线程池处理不过来就疯狂创建新线程从 100 涨到 20000CPU 被线程调度耗尽实际业务处理趋近于 0。每次重启后所有客户端同时重连必然触发这个风暴——形成重启 → 风暴 → 崩溃 → 重启的死循环。修复限制 NIO 线程池在 setenv 的 JVM 参数中加三行-Dorg.apache.activemq.transport.nio.SelectorPool.corePoolSize8-Dorg.apache.activemq.transport.nio.SelectorPool.maximumPoolSize32-Dorg.apache.activemq.transport.nio.SelectorPool.workQueueCapacity2000效果重连风暴来临时NIO 线程最多 32 个超出的连接在 2000 容量的队列里排队。队列满了新连接被maximumConnections拒绝而不是无限创建线程拖垮进程。NIO 的设计就是少量线程处理大量连接。3 万连接 / 8 个核心线程 每线程 3750 连接这是 NIO 的正常工作模式。32 的上限对于 4 核 CPU 已经很充裕了。七、监控脚本的进化整个过程中监控脚本也迭代了好几版记录一下踩过的坑坑1CPU 读数为 0v3 版本用CPU$(get_cpu)赋值但$(...)是子 shell里面修改的全局变量_PREV_TICKS、_PREV_SEC传不出来导致 CPU 计算永远是 0。修复直接调用函数不用$()函数里改全局变量调用者直接读。坑2grep -c || echo 0输出两行COUNT$(grep-cpatternfile||echo0)当 grep 匹配 0 次时grep -c输出0然后 exit 1触发|| echo 0结果变量里是0\n0两行后面的整数比较直接报错。修复改成grep -c pattern file || true再用safe_num函数处理。坑3ss命令缺-ass -Ht -n只统计 ESTABLISHED 状态的连接TIME_WAIT、CLOSE_WAIT 都漏掉了连接数统计少了一大截。修复加-a参数。坑4fork 太多OOM 时自己先挂v3 版本每 30 秒要 fork 15 次ss、grep、ps、free 等高峰期内存 25MB。系统 OOM 时监控脚本因为频繁 fork 先被 kill反而起不到监控作用。修复 v4能读/proc就不用命令。CPU、RSS、线程数、系统内存全部用 bash 读/proc文件0 fork。连接数用 awk 读/proc/net/tcp1 fork。每周期从 15 fork 降到 2 fork内存从 25MB 降到 1MB。坑5重启后不验证插件监控脚本触发重启后只检查 PID 是否存在不验证插件是否加载。曾经出现过重启后配置丢失、插件没加载的情况监控脚本以为启动成功实际系统在裸奔。修复重启后 grep 插件日志确认加载成功。八、最终状态指标故障峰值优化后稳定值连接数45,529~30,000线程数27,231130-400CPU359%26-60%Stealing link (ios-print)~7次/秒0插件拦截Timer already cancelled大量0进程崩溃频率每53分钟一次0生效的配置清单activemq.xml (mqttnio, URI 一行内)maximumConnections30000wireFormat.maxInactivityDuration6000060秒transport.closeIdleConnectionTimeout60000transport.keepAliveTime15000transport.closeAsynctrueallowLinkStealingfalseRejectDuplicateClientIdPlugin插件拦截 ios-printsetenv (JVM 参数)-Xms2048M -Xmx2048M-XX:UseG1GC -XX:MaxGCPauseMillis200-XX:HeapDumpOnOutOfMemoryErrorSelectorPool.corePoolSize8SelectorPool.maximumPoolSize32SelectorPool.workQueueCapacity2000log4j2.propertiesTransportConnection → ERROR抑制拦截连接日志TransportConnector → ERROR抑制连接超限日志MQTTProtocolConverter → ERROR抑制 Broken pipe 堆栈九、经验教训参数是否生效看日志不要猜。Listening for connections是最可靠的验证方式。URI 跨行这种坑不看日志永远发现不了。MQTT clientId 唯一是底线。共用 clientId 等于自毁服务端再怎么优化都只是缓解。客户端修复每台设备生成唯一 clientId才是根本解。NIO 线程池一定要加上限。默认无上限在高并发场景下就是自杀按钮。对于 4 核机器32 个 NIO 线程绰绰有余。监控脚本本身也是系统的一部分。OOM 时监控脚本不能先挂要用最少的资源完成采集。bash 读/proc比调用命令靠谱得多。每加一个 fix 都要验证副作用。比如加插件后要确认日志量不会爆炸改线程池后要确认吞吐不受影响。回滚脚本和部署脚本同样重要。每次部署脚本都配对应的回滚脚本出问题能 30 秒回滚比什么都强。附相关文件索引文件说明RejectDuplicateClientIdPlugin.java自定义插件源码deploy_reject_plugin.sh插件一键编译部署脚本rollback_reject_plugin.sh插件回滚脚本monitor_activemq_v4.shv4 监控脚本OOM 安全轻量版activemq.xml最终版配置URI 单行MqttClient_fixed.cs等客户端修复代码C#