1. 项目概述为什么 MQTT 是 SpringBoot 物联网场景里绕不开的“呼吸系统”你有没有遇到过这样的情况用 SpringBoot 写了个设备管理后台前端页面刷新一次才看到温湿度变化或者给智能插座发个“开机”指令等了三秒才响应用户已经点开另一个App了这不是代码写得烂而是通信协议选错了——HTTP 请求-响应模型天生不适合设备间低延迟、高并发、弱网络下的双向实时交互。而 MQTT就是专为这种场景设计的轻量级发布/订阅消息协议。它不依赖长连接维持靠极小的报文头最小仅2字节、QoS分级机制、遗嘱消息Last Will和会话保持Clean Session四大特性在嵌入式芯片、4G/NB-IoT模组、边缘网关上跑得比 HTTP 轻快十倍。我去年在某高校实验室做的一个农业大棚监控系统87个传感器节点全用 MQTT 接入 SpringBoot 后端单台 2核4G 的云服务器稳定支撑日均 320 万条消息吞吐CPU 峰值从未超过 45%。这背后不是靠堆硬件而是协议层就做了减法MQTT 报文没有 HTTP 那套 Cookie、Header、MIME 类型的冗余字段一条温度上报消息原始数据可能就 12 字节加上协议头也才 16 字节而同等功能的 HTTP POST 请求光是请求行Headers 就干掉 200 字节。SpringBoot 对 MQTT 的支持核心在于spring-integration-mqtt和spring-boot-starter-integration这两个模块它们把底层 Paho 客户端封装成 Spring 生态里熟悉的 MessageChannel、MessageHandler、ServiceActivator 等抽象让你不用碰 TCP 连接管理、心跳重连、线程池调度这些脏活累活。这篇文章不讲“MQTT 是什么”的教科书定义只聚焦一件事如何在 SpringBoot 工程里干净利落地实现主题Topic级别的消息发布与订阅且能应对真实产线环境里的断网重连、消息堆积、QoS 不匹配、主题通配符误用等典型问题。适合正在做物联网平台、工业数采、智能硬件中台的后端开发者也适合刚从 Web 开发转战 IoT 领域、对消息中间件还停留在 RabbitMQ/Kafka 认知的同学——你会发现MQTT 的使用逻辑和你熟悉的 REST API 完全不同它不是“调接口”而是“守候在某个频道里听广播”。2. 整体架构设计与方案选型为什么不用 WebSocket 或自研 TCP 协议2.1 为什么必须用 MQTT而不是“自己造轮子”有人会问既然 SpringBoot 支持 WebSocket我直接用它推消息不行吗或者更激进点用 Netty 自己写个二进制协议这两种思路在小规模 Demo 里确实能跑通但一到真实场景就露馅。我拿去年踩过的一个坑举例某公司想用 WebSocket 实现“远程控制空调”前端页面建立连接后后端通过 session 找到对应设备通道发指令。上线一周后运维报警——WebSocket 连接数暴涨到 1.2 万但实际在线设备只有 3800 台。查日志发现90% 的连接是“假在线”手机切后台、WiFi 切 4G、浏览器标签页休眠都会让 WebSocket 连接处于半死不活状态服务端却还在傻等心跳包。MQTT 的 Clean Session 机制天然解决这个问题客户端断开时Broker比如 EMQX 或 Mosquitto会自动清理其订阅关系和未确认消息下次重连用新会话 ID干净利落。而 WebSocket 没有标准的会话生命周期管理全靠你自己写超时检测、心跳踢出、连接池回收代码量翻倍bug 率飙升。再看自研协议某团队曾用 Netty 开发了一套私有二进制协议宣称“比 MQTT 快 20%”。结果上线三个月光是处理不同厂商设备的时钟漂移、乱序重传、ACK 丢包就改了 17 个版本。MQTT 的 QoS 0/1/2 三级保障是经过全球数亿设备验证的成熟方案——QoS 0 是“发完即忘”适合传感器上报温湿度这类可丢失数据QoS 1 是“至少一次”带 ACK 重传适合下发控制指令QoS 2 是“恰好一次”用四步握手保证不重复不丢失适合金融级设备固件升级。你真有必要为省那几毫秒去挑战十年沉淀的工业标准吗2.2 Broker 选型EMQX vs Mosquitto别被“开源免费”忽悠SpringBoot 本身不提供 MQTT Broker必须外接。主流选择是 EMQX企业级和 Mosquitto轻量级。很多人第一反应选 Mosquitto觉得“小而美”、“Docker 一键启动”。但我在三个不同规模项目里实测下来它的短板非常明显单机最大连接数官方标称 10 万但实测在 3.5 万连接时CPU 就开始抖动内存泄漏明显集群模式需要额外部署 Redis 做共享状态配置复杂度陡增。而 EMQX 5.x 版本单节点轻松扛住 50 万连接集群自动发现、负载均衡、规则引擎SQL 式消息路由都是开箱即用。最关键的是它的可观测性内置 Dashboard 实时显示每个 Topic 的消息速率、延迟分布、客户端在线列表甚至能按 IP 段限流。某次我们发现某批次传感器固件有 Bug疯狂重连刷 TopicEMQX 的“连接频率限制”策略 30 秒内就自动熔断了异常 IP避免拖垮整个集群。反观 Mosquitto你得自己写脚本解析日志、用 Prometheus Grafana 搭监控等发现问题服务已经雪崩了。所以我的建议很直接开发测试阶段用 Mosquitto 快速验证逻辑一旦进入联调或预发布立刻切到 EMQX 社区版完全免费。它的 Docker Compose 启动命令就一行docker run -d --name emqx -p 1883:1883 -p 8081:8081 -p 8083:8083 -p 8084:8084 -p 18083:18083 -e EMQX_LOADED_PLUGINSemqx_auth_username,emqx_management emqx/emqx:5.7.2。其中 18083 是管理后台端口浏览器打开就能看到所有连接和 Topic 流量图。2.3 SpringBoot 集成方式为什么弃用原生 Paho拥抱 Spring Integration早期 SpringBoot 项目常用org.eclipse.paho:org.eclipse.paho.client.mqttv3这个原生 Java 客户端手动 new MqttClient、setCallback、connect、subscribe。这种方式的问题在于它游离于 Spring 容器之外无法享受依赖注入、AOP 切面、事务管理等生态红利。比如你想在消息处理前加个日志切面或者处理失败时触发告警事件都得自己写回调包装。而 Spring Integration 提供的MqttPahoMessageDrivenChannelAdapter和MqttPahoMessageHandler本质是把 MQTT 客户端包装成 Spring 的 MessageChannel所有消息流转都走 Spring 的消息总线。这意味着你可以用ServiceActivator注解标记任意方法作为消费者用Transformer做消息格式转换用Filter做业务规则过滤甚至用Splitter把一条 JSON 消息拆成多条子消息并行处理。更重要的是它内置了重连策略当网络中断时不是简单抛异常而是按指数退避Exponential Backoff自动重试重连间隔从 1 秒开始每次翻倍直到 30 秒封顶期间不阻塞主线程。我见过太多项目因为没处理好重连导致设备离线后消息积压重连瞬间涌进几千条旧消息把业务逻辑冲垮。Spring Integration 的recoveryInterval参数就是你的安全阀。3. 核心细节解析与实操要点主题设计、QoS 选择与安全边界3.1 主题Topic不是路径是“消息路由的坐标系”很多新手把 MQTT Topic 当成文件路径起名device/001/temp、device/001/humid觉得结构清晰。这是危险的惯性思维。MQTT Topic 的本质是分层路由键Hierarchical Routing Key它的/符号不表示目录层级而是分隔符用于支持单层通配和#多层通配匹配。比如你订阅device//temp就能收到所有设备的温度消息订阅device/#则收到该前缀下所有消息。但如果设备 ID 是 UUID如device/a1b2c3d4-e5f6-7890-g1h2-i3j4k5l6m7n8/temp就完全失效必须用#而#的性能开销比高 3~5 倍Broker 要做更多字符串匹配。更致命的是#会匹配到你不关心的消息比如device/xxx/config/update如果业务代码没做二次过滤可能误触发配置更新逻辑。我的经验是Topic 设计要遵循“业务语义优先层级精简至上”原则。例如农业大棚项目我们定义 Topic 为farm/{farmId}/sensor/{sensorType}/{deviceId}其中{farmId}是数字编号如 101{sensorType}是枚举值temp/humid/co2{deviceId}是 6 位数字000001。这样既保证通配可用如farm/101/sensor/temp/又避免 UUID 带来的长度和性能问题。另外绝对禁止在 Topic 中放敏感信息不要用user/12345/password/reset这种设计MQTT Broker 日志、监控面板、网络抓包都能看到 Topic 名密码重置指令应该放在 Payload 里用 AES 加密。3.2 QoS 选择不是越高越好而是“够用即止”QoSQuality of Service是 MQTT 的灵魂但新手常陷入“QoS 2 最安全必须全用”的误区。真相是QoS 级别越高通信开销越大延迟越长设备耗电越多。QoS 0 的报文只有固定头Payload无 ACKQoS 1 多了 PUBACK 报文往返一次QoS 2 要 PUBREC、PUBREL、PUBCOMP 三次握手延迟增加 300%。我做过对比测试在 4G 网络下发送一条 50 字节的温湿度 JSONQoS 0 平均耗时 42msQoS 1 是 118msQoS 2 达到 356ms。对传感器上报这类“可丢失”数据QoS 0 完全足够——丢了就丢下一秒再报业务无感知。但下发“关闭阀门”指令就必须用 QoS 1确保指令至少到达一次即使网络抖动Broker 也会重发直到收到 PUBACK。QoS 2 只适用于极少数场景比如银行 ATM 固件升级要求“恰好一次”不能多也不能少。SpringBoot 里设置 QoS 很简单在MqttPahoMessageHandler配置里指定setQos(1)即可。但要注意QoS 是客户端和服务端协商的结果取双方声明的较低值。如果你的设备端只支持 QoS 1你后端设 QoS 2最终生效的还是 QoS 1。所以务必在设备接入规范里明确标注各 Topic 的推荐 QoS避免“后端设 2设备只认 1”导致消息不可靠。3.3 安全不是“加个用户名密码”就完事MQTT 默认是明文传输用户名密码也是 Base64 编码抓包就能看到。很多项目只在application.yml里配username: admin、password: 123456以为加了认证就安全了。这是严重误区。真正的安全防线有三层传输层加密TLS、应用层认证Username/Password、权限控制ACL。EMQX 的 ACL 规则文件etc/acl.conf可以精确到 Topic 级别{allow, {user, device_001}, publish, [device/001/#]}.表示只允许 device_001 用户向自己的 Topic 发布消息禁止订阅其他设备数据。而 TLS 是底线——没有 TLS所有数据包括密码都在裸奔。EMQX 启动时加-e EMQX_LISTENER__SSL__EXTERNAL__KEYFILE/opt/emqx/etc/certs/key.pem参数即可启用。SpringBoot 客户端连接时URL 要从tcp://localhost:1883改成ssl://localhost:8883并配置MqttConnectOptions的setSocketFactory加载你的证书。我见过最惨的案例某项目没配 TLS黑客用 Wireshark 抓到设备上报的 GPS 坐标和电量反向定位到仓库位置三天后设备被批量盗走。安全不是锦上添花是生存红线。4. 实操过程与核心环节实现从零搭建可运行的发布/订阅系统4.1 环境准备与依赖配置避开 Maven 依赖冲突的深坑SpringBoot 项目起步第一步是加依赖。很多人直接抄网上教程加spring-integration-mqtt结果启动报错NoSuchMethodError: org.springframework.integration.support.management.IntegrationManagementConfigurer.setBeanFactory。这是因为 Spring Integration 5.x 和 SpringBoot 2.7 的版本不兼容。正确姿势是用 SpringBoot 官方 starter而非手动引入 Integration 包。在pom.xml中添加dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-integration/artifactId /dependency dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId !-- 显式指定版本避免传递依赖冲突 -- version5.5.18/version /dependency同时application.yml必须配置 MQTT Broker 连接参数spring: mqtt: url: ssl://localhost:8883 username: admin password: emqx123 client-id: springboot-server # 连接超时单位毫秒 connection-timeout: 30000 # 心跳间隔单位秒 keep-alive-interval: 60 # 是否开启 Clean Session生产环境建议 false保留会话 clean-session: false这里有个关键点clean-session: false。很多教程写成true说“避免消息堆积”。但真实场景中设备可能离线数小时你希望它重连后能收到离线期间的控制指令。EMQX 的会话消息队列Session Queue默认保存 1000 条可通过zone.external.max_mqueue_len 5000在etc/emqx.conf中调大。如果设clean-session: true设备重连时会话清空所有离线消息永久丢失。4.2 消息发布Publisher实现如何避免“发出去就失联”的尴尬发布消息看似简单但隐藏着两个高频陷阱Payload 序列化错误和异步发送阻塞。先看序列化MQTT Payload 是 byte[]你不能直接 send 一个MapString, Object必须序列化。常见错误是用JSON.toJSONString(map)但中文会变成 Unicode 转义\u4f60\u597d设备端解析失败。正确做法是用Jackson2JsonObjectMapper并设置 UTF-8 编码Bean public MessageHandler mqttOutbound() { MqttPahoMessageHandler handler new MqttPahoMessageHandler(tcp://localhost:1883, springboot-publisher); handler.setAsync(true); // 关键设为异步避免阻塞主线程 handler.setDefaultTopic(device/001/control); // 配置 Jackson 序列化器支持中文 ObjectMapper mapper new ObjectMapper(); mapper.configure(JsonGenerator.Feature.WRITE_CHAR_ARRAYS_AS_JSON_ARRAYS, true); handler.setPayloadSerializer(new Jackson2JsonObjectMapper(mapper)); return handler; }handler.setAsync(true)是救命设置。如果不设每次send()都会同步等待 Broker 返回 PUBACKQoS 1或直接返回QoS 0在高并发下线程池迅速耗尽。设为异步后send()立即返回失败时通过setFailureChannel()指定错误处理 Channel。我在线上环境见过因没设异步1000 QPS 下线程数飙到 2000GC 频繁服务直接卡死。4.3 消息订阅Subscriber实现从“收到消息”到“可靠处理”的完整链路订阅的核心是MqttPahoMessageDrivenChannelAdapter但它只是“收音机”真正干活的是后面的ServiceActivator。下面是一个生产环境可用的完整订阅链路Configuration public class MqttConfig { Bean public MessageChannel mqttInputChannel() { return MessageChannels.publishSubscribe().get(); } Bean public MqttPahoMessageDrivenChannelAdapter inbound() { MqttPahoMessageDrivenChannelAdapter adapter new MqttPahoMessageDrivenChannelAdapter(tcp://localhost:1883, springboot-subscriber); adapter.setClientId(springboot-subscriber); adapter.setCompletionTimeout(30000); adapter.setConverter(new DefaultMQTTMessageConverter()); // 自动转 String // 订阅多个 Topic支持通配符 adapter.setTopic(device//temp, device//humid, device//co2); // 设置重连策略初始延迟1秒最大延迟30秒乘数2.0 ExponentialBackOffPolicy backOff new ExponentialBackOffPolicy(); backOff.setInitialInterval(1000); backOff.setMaxInterval(30000); backOff.setMultiplier(2.0); adapter.setRecoveryInterval(1000); adapter.setBackOffPolicy(backOff); adapter.setOutputChannel(mqttInputChannel()); return adapter; } Bean ServiceActivator(inputChannel mqttInputChannel) public MessageHandler handler() { return message - { try { // 1. 解析 Topic 获取设备ID和类型 String topic (String) message.getHeaders().get(mqtt_receivedTopic); String[] parts topic.split(/); String deviceId parts[1]; String sensorType parts[2]; // 2. 解析 Payload 为 JSON String payload (String) message.getPayload(); JsonNode node new ObjectMapper().readTree(payload); double value node.get(value).asDouble(); // 3. 业务处理存数据库、触发告警等 sensorService.saveReading(deviceId, sensorType, value); // 4. 手动 ACKQoS 1/2 必须 if (message.getHeaders().containsKey(mqtt_deliveryAttempt)) { // 这里可以记录日志但无需手动 ACKSpring Integration 已处理 } } catch (Exception e) { log.error(处理 MQTT 消息失败topic{}, payload{}, message.getHeaders().get(mqtt_receivedTopic), message.getPayload(), e); // 重要不要 throw 异常否则消息会被重复投递 // 正确做法是记录错误继续处理下一条 } }; } }这段代码的关键细节setTopic()支持数组一次订阅多个 Topic比循环创建多个 Adapter 节省内存ExponentialBackOffPolicy确保网络抖动时重连不打满 CPUServiceActivator方法里绝不能 throw 异常否则 Spring Integration 会认为处理失败触发重试QoS 1或丢弃QoS 0造成消息重复或丢失。所有异常必须 try-catch 后记录日志保证方法正常返回mqtt_receivedTopic是 Spring Integration 自动注入的 Header无需自己解析原始 MQTT 包。4.4 遗嘱消息Last Will实战让设备“临终托付”成为可能遗嘱消息是 MQTT 的神来之笔当设备异常断开非主动 disconnectBroker 会自动向指定 Topic 发布一条预设消息通知系统“该设备已离线”。这比心跳检测快 5~10 秒。SpringBoot 客户端设置遗嘱非常简单在MqttConnectOptions中配置Bean public MqttConnectOptions mqttConnectOptions() { MqttConnectOptions options new MqttConnectOptions(); options.setUserName(device_001); options.setPassword(pass123.toCharArray()); options.setCleanSession(false); // 设置遗嘱消息 options.setWill(device/status, offline.getBytes(), 1, true); // 遗嘱 Topic: device/status // 遗嘱 Payload: offline 字节数组 // QoS: 1确保送达 // Retained: true保留消息新订阅者立即收到 return options; }注意setWill()的第四个参数retained设为true。这样当设备断开Broker 会把offline作为保留消息Retained Message发布到device/status。任何新订阅该 Topic 的客户端会立即收到这条“最后状态”无需等待下一次设备上线。我们在大棚系统里用这个特性实现了“设备健康看板”前端页面订阅device/status一打开就看到所有设备当前在线/离线状态不用轮询接口。遗嘱消息的 Payload 建议用 JSON 格式包含时间戳和原因比如{status:offline,ts:1712345678,reason:network_timeout}方便后续分析离线根因。5. 常见问题与排查技巧实录那些文档里不会写的“血泪教训”5.1 问题现象消息“发出去了”但订阅端永远收不到这是新手最高频问题。表面看代码没问题send()也不报错但另一端就是没动静。排查必须按顺序来确认 Broker 状态docker ps | grep emqx看容器是否运行curl http://localhost:18083/api/v4/brokers查 API 是否返回 JSON检查 Topic 拼写MQTT 区分大小写Device/001/temp和device/001/temp是两个 Topic。用 EMQX Dashboard 的 “WebSocket Client” 工具手动 publish 一条消息到目标 Topic看订阅端能否收到——这能快速定位是代码问题还是 Broker 配置问题验证订阅关系在 EMQX Dashboard 的 “Clients” 页面找到你的客户端 ID如springboot-subscriber点进去看 “Subscriptions” 列表确认是否成功订阅了目标 Topic。如果为空说明setTopic()没生效检查MqttPahoMessageDrivenChannelAdapterBean 是否被正确创建抓包验证tcpdump -i any port 1883 -w mqtt.pcap抓包用 Wireshark 打开过滤mqtt看是否有 SUBSCRIBE 报文发出以及 Broker 是否返回 SUBACK。没有 SUBSCRIBE说明客户端根本没发起订阅请求。提示Spring Integration 的MqttPahoMessageDrivenChannelAdapter默认日志级别是 INFO看不到详细连接过程。在logback-spring.xml中加logger nameorg.springframework.integration.mqtt levelDEBUG/启动时就能看到 “Subscribing to topics: [device/001/temp]” 这类日志比抓包快十倍。5.2 问题现象设备重连后收到大量“历史消息”业务逻辑被冲垮根源在于clean-session: true和会话消息队列Session Queue的叠加效应。当设备以clean-session: true重连Broker 会为其创建新会话但旧会话的队列消息不会自动清除而是堆积在 Broker 内存中。新会话建立后Broker 会把旧队列消息一股脑推给新会话。解决方案有三首选clean-session: false并在设备端实现“消息去重”。给每条消息加唯一 ID如 UUID服务端用 Redis Set 记录已处理 ID10 分钟过期次选在 EMQX 中配置zone.external.max_mqueue_len 10把会话队列长度限制为 10 条超出的自动丢弃应急在ServiceActivator方法开头加时间判断if (System.currentTimeMillis() - messageTimestamp 60000) return;丢弃 1 分钟前的消息。5.3 问题现象QoS 1 消息重复消费数据库出现双写QoS 1 的“至少一次”语义意味着网络不稳定时Broker 可能重发 PUBREL导致同一条消息被消费两次。这不是 Bug是协议设计使然。解决办法只有一个业务逻辑必须幂等。比如“增加设备在线时长”不要用UPDATE device SET online_time online_time ?而要用UPDATE device SET online_time ? WHERE last_update_ts ?用时间戳做乐观锁。或者更简单用 Redis 的SETNX命令以消息 ID 为 key设置成功才执行业务失败直接 return。我见过最优雅的方案是“状态机驱动”设备上报消息带seq_no序列号服务端用Redis ZSET存储该设备最近 100 个已处理 seq_no每次处理前ZSCORE查询存在则跳过。5.4 问题现象EMQX Dashboard 显示连接数正常但消息吞吐量为 0这通常是因为ACL 权限配置错误。EMQX 的 ACL 规则文件etc/acl.conf是白名单机制默认拒绝所有操作。如果你只写了allow, {user, admin}, publish, [#]但没写subscribe权限那么该用户只能发消息不能收消息。检查方法在 Dashboard 的 “Clients” 页面找到客户端看 “Permissions” 列是否显示publish, subscribe。修复只需在acl.conf中加一行{allow, {user, springboot-subscriber}, subscribe, [device//temp]}.。记住publish 和 subscribe 权限必须分开配置不能混为一谈。5.5 问题现象SpringBoot 启动时报MqttException: MqttException (0) - java.net.ConnectException: Connection refused这看似是网络问题但 90% 的原因是EMQX 的监听器配置未开启 SSL。你在application.yml里写了ssl://localhost:8883但 EMQX 默认只监听1883TCP和1883WebSocket8883SSL端口需要显式开启。编辑etc/emqx.conf取消注释并修改listener.ssl.external 8883 listener.ssl.external.keyfile etc/certs/key.pem listener.ssl.external.certfile etc/certs/cert.pem然后重启 EMQX。生成自签名证书的命令是openssl req -x509 -nodes -days 365 -newkey rsa:2048 -keyout key.pem -out cert.pem。证书路径必须和配置文件中一致且cert.pem和key.pem文件权限设为600否则 EMQX 启动失败。注意SpringBoot 客户端连接 SSL 时如果证书是自签名的必须禁用证书校验否则javax.net.ssl.SSLHandshakeException。在MqttConnectOptions中加options.setSocketFactory(SSLSocketFactory.getDefault());即可生产环境务必换成受信任 CA 签发的证书。6. 实战扩展与进阶技巧让 MQTT 能力不止于“发消息”6.1 用 EMQX 规则引擎实现“消息自动路由”减少 SpringBoot 代码量EMQX 5.x 内置 SQL 式规则引擎能直接在 Broker 层做消息过滤、富化、转发把 SpringBoot 从“消息搬运工”解放出来。比如你想把所有device//temp消息当温度 35℃ 时自动转发到alarm/high-tempTopic并附带设备位置信息。在 EMQX Dashboard 的 “Rule Engine” 页面创建规则SELECT payload.temp as temperature, payload.device_id as device_id, Shanghai as location, timestamp() as ts FROM device//temp WHERE payload.temp 35动作选择 “Data Bridge → MQTT Bridge”目标 Broker 填localhost:1883目标 Topic 填alarm/high-temp。这样SpringBoot 只需订阅alarm/high-temp就能拿到预处理好的告警消息不用在代码里写 if 判断。规则引擎还支持调用 Webhook把消息直接 POST 到 SpringBoot 的 REST 接口实现“MQTT 与 HTTP 的无缝桥接”。6.2 用 Spring Integration 的Splitter处理批量上报消息某些设备如 LoRa 网关会把多个传感器数据打包成一条 JSON 上报例如{ gateway_id: gw-001, readings: [ {sensor_id: s001, type: temp, value: 25.3}, {sensor_id: s002, type: humid, value: 65.2} ] }如果用传统方式你得在ServiceActivator里手动遍历readings数组逐条处理。而 Spring Integration 的Splitter能自动拆分Bean Splitter(inputChannel mqttInputChannel, outputChannel splitOutputChannel) public GenericTransformerMessage?, ListMessage? splitter() { return message - { JsonNode node new ObjectMapper().readTree((String) message.getPayload()); ArrayNode readings (ArrayNode) node.get(readings); ListMessage? messages new ArrayList(); for (JsonNode reading : readings) { MapString, Object payload new HashMap(); payload.put(gateway_id, node.get(gateway_id).asText()); payload.put(sensor_id, reading.get(sensor_id).asText()); payload.put(type, reading.get(type).asText()); payload.put(value, reading.get(value).asDouble()); messages.add(MessageBuilder.withPayload(payload).copyHeaders(message.getHeaders()).build()); } return messages; }; }拆分后的每条消息会进入splitOutputChannel再由下游的ServiceActivator逐条处理。这样代码更清晰也便于对单条消息做独立重试。6.3 用Transformer统一处理消息格式避免“每个 Handler 都写解析”不同设备厂商上报的 JSON 格式千奇百怪有的叫temperature有的叫temp_val有的时间戳是ts有的是timestamp。如果每个ServiceActivator都写一遍node.get(xxx)维护成本爆炸。Transformer就是标准化入口Bean Transformer(inputChannel mqttInputChannel, outputChannel transformedChannel) public GenericTransformerMessage?, Message? transformer() { return message - { String payload (String) message.getPayload(); JsonNode node new ObjectMapper().readTree(payload); // 统一映射到标准字段 MapString, Object standard new HashMap(); standard.put(device_id, node.path(device_id).asText()); standard.put(sensor_type, node.path(type).asText(node.path(sensor_type).asText())); standard.put(value, node.path(value).asDouble(node.path(temp).asDouble(0.0))); standard.put(timestamp, node.path(ts).asLong(node.path(timestamp).asLong(System.currentTimeMillis()))); return MessageBuilder.withPayload(standard) .copyHeaders(message.getHeaders()) .build(); }; }所有消息经过transformer后Payload 都是统一的MapString, Object下游 Handler 只需payload.get(device_id)彻底告别字段名战争。7. 我在实际项目中的几个关键体会这个项目做完回看有几个点让我现在想起来还觉得庆幸。第一个是坚持用 EMQX 而不是 Mosquitto。当时团队有同事说“Mosquitto 更轻量资源占用少”差点说服我。但上线前我做了压力测试用 JMeter 模拟 5 万设备并发连接Mosquitto 的内存增长曲线像坐火箭而 EMQX 稳定在 1.2G。第二个是