面对“设备上报延迟高、后台下发指令要靠轮询”这类老问题我最初也没想到用MQTT直到有一次做设备管理后台才真正体会到SpringBoot整合MQTT带来的实时性红利。这篇文章是我把整套流程跑通之后整理出来的完整笔记覆盖依赖引入、配置项拆解、连接管理、订阅回调、发布消息以及我在实测中踩过的五个典型坑和完整排查思路希望能帮到刚开始接触MQTT集成、又担心“资料太散、照抄跑不通”的Java后端开发同学。1. 先想清楚为什么非要用MQTT而不是HTTP轮询很多团队做设备指令下发第一反应是“写个REST接口让设备定时拉取”。这个方案在设备量小、实时性要求不高的时候确实能跑但我亲自做过一个告警推送服务后发现轮询模式在真实场景里有三个非常难受的地方第一设备要维持实时性轮询间隔就得压到几秒钟请求量直接翻好几倍服务端压力很大第二弱网环境下设备频繁建连拉取功耗和流量都扛不住第三服务端想主动通知某个设备时轮询永远有延迟指令下发做不到真正的“即发即达”。1.1 四种常见方案摆在桌面上对比我平时给人讲技术选型喜欢先列一张对比表把HTTP轮询、WebSocket、自研TCP长连接和MQTT放到一起看优劣一眼就清楚。方案实时性弱网表现实现成本服务端压力典型场景HTTP轮询差取决于间隔差频繁建连低高低频率状态上报WebSocket好一般断线重连靠自研中中浏览器实时推送自研TCP长连接好取决于实现很高中游戏、自有协议MQTT好好协议本身为弱网设计低低物联网设备通信、即时消息表格里的“实现成本低”初看可能觉得不太直观。其实MQTT对后端开发者来说最大的友好之处在于发布、订阅、QoS这些概念都已经被协议层实现好了我们只需要关心业务消息本身不用去处理粘包拆包、心跳维护、断线重连这些底层难题。1.2 MQTT到底解决了哪三类问题第一类是弱网环境下的可靠性。MQTT基于TCP但它在协议层面设计了重连机制、会话延续和消息QoS等级设备端网络抖一抖重新连上之后还能继续收离线期间的消息这是HTTP轮询很难做到的。第二类是消息过滤与分发。设备端按主题订阅服务端往固定主题发布消息Broker会负责把消息精准推给订阅者。发布者不需要知道谁在订阅订阅者也不需要知道消息从哪里来这对系统解耦帮助很大。第三类是低功耗场景。MQTT的连接保持机制很轻量心跳报文很小设备不需要为了“等一下可能有一条消息”而反复建立连接。我做过的几个设备项目里同样频率的心跳上报用MQTT比用HTTP轮询省电明显。1.3 适合哪类读者学习路径怎么走这篇文章适合已经会用SpringBoot写普通接口、但对消息协议接触不多的后端同学。建议你先别急着看代码先把“发布者-代理-订阅者”这个三角关系在脑子里过一遍服务端既可以作为订阅者接收设备上报也可以作为发布者向设备下发指令。理解了这个模型后面所有代码都是在给这个模型落地。2. 环境与依赖版本搭配决定后续是否顺利SpringBoot整合MQTT我踩过的最早一个坑就是版本搭配。网上能找到的教程五花八门有的基于SpringBoot 1.x有的基于2.x还有的推荐直接引入Spring Integration MQTT模块。如果你照着一个老教程的依赖坐标硬抄启动时大概率会碰到ClassNotFound或者Bean冲突。2.1 版本选型我最终采用的组合我本地和测试环境长期使用的组合是SpringBoot 2.7.x JDK 8/11 Java MQTT客户端库。为什么不用SpringBoot 3.x不是不能用而是3.x基于Jakarta命名空间一些老依赖需要同步升级很多教程和网上的片段还没跟上新手容易把自己绕晕。另外我推荐直接使用Java生态里最常用的MQTT客户端库而不是Spring Integration的封装模块。Spring Integration MQTT确实能提供MessageHandler之类的抽象但它把很多细节藏起来了出了问题不好排查。自封装客户端库虽然多写几个类但每个连接参数都在自己手里线上查问题的时候会舒服很多。2.2 Maven依赖引入与说明在pom.xml里加入下面的依赖dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependency这里用到的版本是目前社区里最稳定、兼容性最好的Java版MQTT客户端库之一。如果你所在团队的私服没有缓存这个版本可以先让Maven重新拉取一次或者临时换成1.2.2功能差别不大但建议线上统一版本避免后续维护混乱。依赖其实就这一个核心的。SpringBoot自带的spring-boot-starter本身不需要额外配置Jackson之类的也不需要因为MQTT消息体默认就是字节数组我们自己决定怎么反序列化。2.3 本地快速搭一个测试用的Broker整合MQTT不能没有一个消息代理我建议你本地用Docker起一个轻量级的MQTT服务端镜像总共就一条命令的事情。docker run -d -p 1883:1883 --name local-mqtt eclipse-mosquitto:2这个容器启动后本地1883端口就是一个完整的MQTT Broker默认是匿名访问。我用这种感觉无压力的环境把代码跑通之后再丢到测试服务器的正式代理上效率高很多。如果你的机器没有Docker也可以安装一个开源的PC端MQTT代理程序Windows和Linux都有对应版本原理一样。启动Broker之后可以用命令行客户端做连通性测试也可以在后面章节的代码写好之后直接用Java客户端连上来验证。注意本地测试Broker的默认配置允许匿名连接但到了正式环境一定要配置账号密码这一点我会在后面配置章节详细说。3. application.yml配置逐项拆解每个字段都是一个坑以前看一些教程配置文件直接贴一坨注释都不写几个看得人一头雾水。我自己把配置拆开逐项解释清楚后才发现很多“连不上”“收不到消息”的问题根源就在配置字段的语义没搞清楚。3.1 一份完整可用的配置模板我习惯把这些参数统一放在以mqtt开头的自定义配置下方便统一管理和加密处理。下面是经过线上验证的一份配置模板可以直接改造使用spring: application: name: mqtt-service mqtt: broker-url: tcp://127.0.0.1:1883 client-id: service-backend-01 username: device-user password: device-pass clean-session: false keep-alive-interval: 60 connection-timeout: 30 auto-reconnect: true topics: - name: device//report qos: 1 - name: gateway/alarm qos: 2这里把topics设计成列表是因为实际项目中通常要同时订阅多个主题而且不同主题的QoS等级可能还不一样。在配置里集中维护比散落在代码里好改得多。3.2 关键字段含义和推荐值配置字段含义推荐值注意事项broker-urlBroker地址tcp://ip:1883SSL场景用ssl://并配证书client-id客户端唯一标识服务名序号同一Broker下必须全局唯一username连接账号Broker中创建的用户匿名测试环境可不填password连接密码强密码加密存储勿明文硬编码clean-session会话清理标志false影响离线消息是否保留keep-alive-interval心跳间隔秒数60需小于网关空闲断开时间connection-timeout连接超时秒数30弱网可适当调大auto-reconnect是否自动重连true正式环境建议加手动重试策略topics订阅主题集合按业务设计支持通配符和#3.3 clientId和cleanSession的关系为什么收不到离线消息这一节是重点几乎每个刚接触MQTT的人都会在这里卡住。我先说结论如果你把clean-session设成true客户端断开后Broker会立即清掉这个客户端的会话信息和所有未消费消息那么离线期间发给它的消息等它重新连上来的时候已经彻底消失了表现就是“消息静悄悄丢了”。反过来把clean-session设成falseBroker会保留客户端的订阅关系和离线消息等客户端重新连接并且用同一个clientId登录时离线期间积压的消息会继续下发。理解了这一点你就不难理解为什么clientId必须保持稳定且唯一。同一个服务实例每次重启连上Broker时如果你换个新的clientId那和之前的订阅关系完全断开了离线消息自然收不到如果两个实例用了同一个clientId则会造成后者把前者踢下线谁都没法稳定工作。所以我在实际配置中会把clientId设计成类似service-name-节点序号的形式每个节点一个固定值重启不换节点之间不重复。设备和后台服务都是这个规则。4. 核心代码连接管理、订阅、回调与发布的完整实现配置到位后代码就是整个集成的骨架。我不喜欢把一堆逻辑堆在一个类里而是分成配置类、连接管理器、订阅回调、发布服务几个部分每个部分只干一件事出问题的时候定位非常快。4.1 配置类把YAML变成强类型对象SpringBoot的配置绑定是最基础的能力先写一个配置类把前面application.yml里的参数映射进来。Data Component ConfigurationProperties(prefix mqtt) public class MqttProperties { private String brokerUrl; private String clientId; private String username; private String password; private boolean cleanSession true; private int keepAliveInterval 60; private int connectionTimeout 30; private boolean autoReconnect true; private ListTopicConfig topics new ArrayList(); Data public static class TopicConfig { private String name; private int qos 1; } }ConfigurationProperties的好处是类型安全还能在启动时就校验字段。比如brokerUrl漏配的时候SpringBoot会直接报错总比运行时连不上再去翻日志强。4.2 连接管理器启动即连接对外提供客户端实例连接管理的核心是把MqttClient的生命周期管理好包括初始化连接、获取当前连接状态、重连触发等。我写成一个MqttConnectionManager由Spring容器管理单例。Component public class MqttConnectionManager { private static final Logger log LoggerFactory.getLogger(MqttConnectionManager.class); private final MqttProperties properties; private MqttClient client; private MqttCallback callback; public MqttConnectionManager(MqttProperties properties) { this.properties properties; } public void connect() throws MqttException { if (client null) { synchronized (this) { if (client null) { client new MqttClient(properties.getBrokerUrl(), properties.getClientId(), new MemoryPersistence()); } } } if (!client.isConnected()) { MqttConnectOptions options buildOptions(); client.setCallback(callback); client.connect(options); log.info(MQTT连接成功clientId{}, properties.getClientId()); } } private MqttConnectOptions buildOptions() { MqttConnectOptions options new MqttConnectOptions(); options.setCleanSession(properties.isCleanSession()); options.setKeepAliveInterval(properties.getKeepAliveInterval()); options.setConnectionTimeout(properties.getConnectionTimeout()); options.setAutomaticReconnect(properties.isAutoReconnect()); if (StringUtils.hasText(properties.getUsername())) { options.setUserName(properties.getUsername()); options.setPassword(properties.getPassword().toCharArray()); } return options; } public MqttClient getClient() { return client; } public boolean isConnected() { return client ! null client.isConnected(); } }这里有几个容易被忽略的细节第一new MqttClient的时候要指定MemoryPersistence否则有些环境会持续写小文件时间久了会出现奇怪的磁盘占用第二setAutomaticReconnect(true)只对客户端断开后的自动重连有效如果connect()本身抛异常该选项并不会无限重试需要我们自己配合重试策略第三client.setCallback必须在connect之前设置否则连接建立后一旦有消息到达回调还没注册消息就丢了。4.3 订阅管理连接成功后注册主题连接成功不代表能收到消息还要主动订阅主题。我把订阅动作抽成一个独立方法既可以在启动阶段调用也可以在动态新增主题时复用。public void subscribeTopics() throws MqttException { if (client null || !client.isConnected()) { throw new IllegalStateException(MQTT客户端未连接无法订阅); } for (MqttProperties.TopicConfig topic : properties.getTopics()) { client.subscribe(topic.getName(), topic.getQos()); log.info(订阅主题成功topic{}qos{}, topic.getName(), topic.getQos()); } }这里需要特别说明QoS的选择。业务上都希望消息可靠所以很多人统一设成QoS2这其实是个误区。QoS2虽然最可靠但吞吐量最低Broker和客户端都要多几轮确认报文。我的经验是普通设备上报用QoS1就够告警、命令这类绝对不能丢的消息才用QoS2而实时性要求高、丢了也无所谓的遥测数据可以用QoS0。订阅之后客户端收到消息时会进入回调方法这是下一步的核心。4.4 消息回调解析、处理、容错都要做全回调是MQTT集成的灵魂。我见过太多项目在回调里直接写业务逻辑结果一个消息处理阻塞后面的消息全部排队积压设备端表现为“服务没响应”。我的做法是回调里只做基础解析和线程池投递真正的业务处理丢给独立线程池。Component public class MqttMessageCallback implements MqttCallback { private static final Logger log LoggerFactory.getLogger(MqttMessageCallback.class); private final ExecutorService messageProcessor; public MqttMessageCallback() { this.messageProcessor new ThreadPoolExecutor( 4, 16, 60, TimeUnit.SECONDS, new LinkedBlockingQueue(2000), new ThreadFactoryBuilder().setNameFormat(mqtt-handler-%d).build(), new ThreadPoolExecutor.CallerRunsPolicy()); } Override public void connectionLost(Throwable cause) { log.error(MQTT连接丢失原因{}, cause.getMessage(), cause); } Override public void messageArrived(String topic, MqttMessage message) { byte[] payload message.getPayload(); messageProcessor.execute(() - { try { String content new String(payload, StandardCharsets.UTF_8); log.info(收到消息topic{}content{}, topic, content); handleMessage(topic, content); } catch (Exception e) { log.error(处理MQTT消息异常topic{}, topic, e); } }); } Override public void deliveryComplete(IMqttDeliveryToken token) { // 发布确认事件可在此记录消息已到达Broker } private void handleMessage(String topic, String content) { // 业务方法根据topic路由到不同处理器 } }CallerRunsPolicy这里是个关键细节。当线程池满时新任务会在调用者线程也就是MQTT客户端回调线程里执行虽然会阻塞后续消息接收但至少不会丢消息。对于物联网场景宁可处理慢一点也不能让消息凭空消失。4.5 发布工具类从服务端向设备下发指令服务端下发指令本质上就是“向某个主题发布消息”。我封装了一个MqttPublisher业务方只需要调用publish(topic, payload)即可。Component public class MqttPublisher { private static final Logger log LoggerFactory.getLogger(MqttPublisher.class); private final MqttConnectionManager connectionManager; public MqttPublisher(MqttConnectionManager connectionManager) { this.connectionManager connectionManager; } public void publish(String topic, String payload, int qos) throws MqttException { ensureConnected(); MqttMessage message new MqttMessage(payload.getBytes(StandardCharsets.UTF_8)); message.setQos(qos); message.setRetained(false); connectionManager.getClient().publish(topic, message); log.info(消息发布成功topic{}qos{}payload{}, topic, qos, payload); } private void ensureConnected() throws MqttException { MqttClient client connectionManager.getClient(); if (client null || !client.isConnected()) { connectionManager.connect(); } } }我在实际项目中还会额外加一个带重试的版本publishWithRetry方法里发布失败时最多重试三次间隔1秒、2秒、4秒。为什么要重试因为MQTT发布时如果连接意外断开client.publish会抛MqttException而这时消息并没有真正到达Broker直接吞掉异常会导致指令丢失这对设备控制类业务是不能接受的。4.6 启动编排先建Bean再连接再订阅最后需要一个入口把这些步骤串起来。我用SmartLifecycle而不是PostConstruct因为PostConstruct执行时机在Spring容器完成依赖注入之后、但无法准确控制多个Bean之间的先后顺序。如果连接管理类的Bean还没有初始化完订阅动作可能拿到空客户端。Component public class MqttLifecycle implements SmartLifecycle { private volatile boolean running false; private final MqttConnectionManager connectionManager; public MqttLifecycle(MqttConnectionManager connectionManager) { this.connectionManager connectionManager; } Override public void start() { try { connectionManager.connect(); // connect成功后注册回调并订阅主题 MqttClient client connectionManager.getClient(); client.setCallback(new MqttMessageCallback()); connectionManager.subscribeTopics(); running true; } catch (MqttException e) { log.error(MQTT启动失败, e); } } Override public void stop() { try { if (connectionManager.getClient() ! null) { connectionManager.getClient().disconnect(); connectionManager.getClient().close(); } } catch (MqttException e) { log.error(MQTT关闭异常, e); } running false; } Override public boolean isRunning() { return running; } Override public int getPhase() { return Integer.MAX_VALUE; } }getPhase返回Integer.MAX_VALUE是让这个Lifecycle在应用启动的最后阶段再执行确保依赖的其他Bean都已经准备就绪。如果你觉得这块比较绕直接用ApplicationRunner也可以效果类似都是为了避免Bean初始化顺序踩坑。5. 亲测有效的排坑记录五个典型事故的完整排查链路这部分是我觉得这篇文章最值钱的地方。网上能找到的整合教程不少但很少有人把“现象是什么、怎么排查、根因在哪、怎么解决”完整写出来。下面五个坑都是我实际工作中碰到过的每一个都能对应到具体场景。5.1 clientId互踢两个实例轮流掉线日志里全是“Lost connection”现象服务部署了两个节点启动后过不了多久节点A的日志出现连接丢失紧接着节点B也出现连接丢失两个节点交替着上线、下线整体服务不可用。排查过程我先看Broker日志发现同一个clientId多次注册连接再看应用配置两个节点的配置都是从配置中心拉取的里面写的clientId是同一个固定值。到这里根因已经清楚了。根因MQTT协议要求每个客户端连接使用唯一的clientId。当新连接使用同一个clientId建立时Broker会主动断开旧连接。两个节点互相抢占形成互踢。解决修改clientId生成规则在配置中心为每个节点维护不同的clientId或者启动时在固定clientId后面拼接一个节点标识。我用的是服务名-节点序号方案节点序号通过环境变量注入简单可靠。5.2 订阅主题收不到消息cleanSession true加重启时机问题现象服务A订阅了device//report主题测试时用客户端工具往该主题发消息服务A收不到但另一个订阅同主题的服务B能收到。排查过程我先确认服务A确实订阅成功启动日志有“订阅主题成功”再确认发布用的主题名和订阅主题匹配。随后我检查了服务A的clean-session配置发现是true而且服务A在测试前曾经重启过几次。关键点来了如果服务A重启前已经订阅了消息但重启时cleanSession为trueBroker会清除旧会话服务A重启后再订阅只能收到之后发布的消息。发布动作发生在服务A重启完全完成之前的那几秒消息自然就丢了。根因测试时发送消息的时机早于客户端重新订阅完成的时机加上cleanSessiontrue不会保留离线消息造成“配置都对但消息没收到”的错觉。解决对需要可靠接收的主题把cleanSession改为false同时确保业务消息是在订阅方法返回之后再发布。排查这类问题还有个技巧用两个客户端一个专门订阅一个专门发布先把发布订阅链路本身验证通再来看业务逻辑。5.3 QoS不一致导致消息丢失发布端QoS1订阅端QoS0现象设备端上报数据用QoS1服务端订阅主题时写的也是QoS1但偶发丢消息统计下来丢掉的大多发生在Broker重启或网络抖动后的短时间内。排查过程我第一次查的时候只看了发布端代码确认是QoS1但忽略了订阅时的QoS实际生效规则。MQTT协议里最终投递的QoS等级由发布端和订阅端共同决定两者中较低的那个生效。如果订阅时请求的是QoS0即使发布端用QoS1发布Broker投递给该订阅者时仍然会降级为QoS0消息没有确认机制网络抖动时就可能丢。根因订阅端配置的QoS低于发布端投递等级被压低。解决把订阅端的QoS改到和发布端一致。如果业务上确实允许丢那就明确接受QoS0如果丢不起两端都要配置QoS1或QoS2并且要在代码注释里写清楚防止后续维护的人改乱。5.4 keepAlive与防火墙冲突连接频繁断开又频繁重连现象服务上线后日志显示连接每隔一两分钟就断一次断开后能自动重连但重连后没多久又断。整体服务可用但消息接收延迟大时断时续。排查过程我先怀疑网络不稳定ping Broker所在的服务器延迟和丢包都正常。接着看网关和防火墙配置发现网络设备里配置了空闲连接超时时间为90秒。而我们MQTT客户端的keepAlive设置为120秒也就是说在这个间隔内客户端没有发送任何MQTT心跳包防火墙会把连接当作空闲连接断开。根因keep-alive-interval设置得比网络中间设备的空闲超时时间更长心跳还没发连接就被外部切断。解决把keepAlive改为45秒或60秒小于网关空闲超时时间即可。同时要注意MQTT协议的keepAlive只是“期望的心跳间隔”客户端库会自动在空闲达到一定比例时发送PINGREQ因此这个值必须和网络链路里的各种超时时间匹配。5.5 Spring容器初始化顺序启动时订阅总是报“客户端未连接”现象本地代码跟着教程写完启动SpringBoot应用时日志报错MQTT客户端未连接无法订阅。但奇怪的是单独用单元测试调用connect再subscribe却是正常的。排查过程我先看启动日志里的Bean初始化顺序发现订阅逻辑所在的Bean先于连接管理Bean执行。原因是我在最开始用的PostConstruct方法里直接调了subscribeTopics()那时MqttConnectionManager虽然被注入了但它自己还没有完成connect()所以客户端处于未连接状态。根因Spring容器中Bean的初始化顺序没有显式编排PostConstruct无法保证相对顺序。解决改用4.6节里的SmartLifecycle方式或者用ApplicationReadyEvent监听器在应用完全启动后再执行连接和订阅。不要指望靠Order注解解决所有顺序问题生命周期接口才是为这种场景设计的。6. 上线前建议加的优化重连、通配符、遗嘱消息与线程模型核心流程跑通后项目距离生产上线还差几步。我把这些优化称为“生产环境四件套”缺了任何一个系统都能跑但稳定性会差一个档次。6.1 动态订阅与多主题通配符业务一复杂主题往往不是写死在配置文件里的而是运行时才能确定。比如某个新设备接入就要订阅一个针对该设备控制的专属主题。此时可以用client.subscribe动态增加订阅不需要重启服务。主题命名的和#通配符也要熟练使用。代表单层通配#代表多层通配。例如device//report匹配device/001/report而device/#匹配device/001/report和device/001/status都能匹配。我建议业务设计上尽量把主题层级控制在三层以内层级越多通配符匹配规则越复杂维护成本越高。6.2 遗嘱消息与保留消息让“掉线”可以被感知设备会掉线服务也会掉线这是常态。MQTT的遗嘱消息Last Will就是用来通知其他订阅者“某个客户端掉线了”。在客户端连接时设置遗嘱Broker会在连接异常断开时替该客户端向指定主题发布一条遗嘱消息。MqttConnectOptions options new MqttConnectOptions(); options.setWill(device/status/offline, {\clientId\:\service-backend-01\}.getBytes(StandardCharsets.UTF_8), 1, false);我在设备状态管理项目里的做法是每个设备上线时发布一条device/status/online的保留消息同时连接设置遗嘱主题为device/status/offline。保留消息意味着新订阅者连上来就能立刻拿到最新的状态不必等设备下次上报。这样后台可以实时维护一张设备在线状态表。6.3 回调线程池永远不要让MQTT的回调线程卡在业务处理上读者可以做一个实验在messageArrived里直接加个Thread.sleep(5000)然后连续给Broker发送多条消息观察客户端接收情况。你会发现后面的消息全部在排队实时性几乎归零。这只是单线程阻塞的演示但现实中调用外部服务超时、数据库慢查询、分布式锁等待等问题都会让回调线程卡住。生产环境我一般把核心线程数设定为设备连接数的十分之一到五分之一最大线程数不超过一个合理上限队列容量要给足。线程池拒绝策略我推荐CallerRunsPolicy宁可阻塞回调不要再丢消息。另外要给线程池里的任务加超时控制比如用Future配合get(timeout)防止个别慢任务长期占用线程。6.4 连接状态监控与消息幂等消费消息没有丢失不代表没有重复。MQTT的QoS1和QoS2在特殊情况下都可能导致重复投递。最典型的是网络抖动时客户端收到消息后还没来得及确认连接就断了重连后Broker会认为消息未投递成功再次下发给客户端。所以集成MQTT的消费端必须考虑幂等。我在处理设备消息时会从消息体中提取一个全局唯一的消息ID在Redis里做去重例如SETNX mqtt:msg:设备ID:消息ID 1 EX 600如果设置成功才执行业务逻辑否则直接跳过。连接状态监控方面可以在connectionLost回调里上报指标也可以在定时任务里检查isConnected()状态。如果项目接入了监控系统建议把MQTT的连接状态、消息积压量、线程池活跃度都暴露出去超过阈值直接告警不然线上出了问题很难快速发现是MQTT链路的问题。最后再分享一个小技巧每次我搭建MQTT测试环境都会在本地同时开一个带图形界面的MQTT客户端工具一边用Java代码发布消息一边用图形工具订阅同一个主题这样能直观看到消息流转是否正常。遇到Java客户端收不到消息的诡异问题先别急着改代码用图形工具代替Java端做订阅如果图形工具能收到问题大概率在Java代码配置上如果图形工具也收不到那就回头查Broker权限、主题命名和QoS设置。我在本地调试时还有一个习惯把日志级别调整到DEBUG重点观察connectionLost和messageArrived两个方法。MQTT链路的问题九成都能从这两个方法的关键日志里找到蛛丝马迹。搞懂上面这些配置和坑之后SpringBoot整合MQTT就不再是什么玄学问题了按部就班地搭你也能一次跑通。
面对“设备上报延迟高、后台下发指令要靠轮询”这类老问题我最初也没想到用MQTT直到有一次做设备管理后台才真正体会到SpringBoot整合MQTT带来的实时性红利。这篇文章是我把整套流程跑通之后整理出来的完整笔记覆盖依赖引入、配置项拆解、连接管理、订阅回调、发布消息以及我在实测中踩过的五个典型坑和完整排查思路希望能帮到刚开始接触MQTT集成、又担心“资料太散、照抄跑不通”的Java后端开发同学。1. 先想清楚为什么非要用MQTT而不是HTTP轮询很多团队做设备指令下发第一反应是“写个REST接口让设备定时拉取”。这个方案在设备量小、实时性要求不高的时候确实能跑但我亲自做过一个告警推送服务后发现轮询模式在真实场景里有三个非常难受的地方第一设备要维持实时性轮询间隔就得压到几秒钟请求量直接翻好几倍服务端压力很大第二弱网环境下设备频繁建连拉取功耗和流量都扛不住第三服务端想主动通知某个设备时轮询永远有延迟指令下发做不到真正的“即发即达”。1.1 四种常见方案摆在桌面上对比我平时给人讲技术选型喜欢先列一张对比表把HTTP轮询、WebSocket、自研TCP长连接和MQTT放到一起看优劣一眼就清楚。方案实时性弱网表现实现成本服务端压力典型场景HTTP轮询差取决于间隔差频繁建连低高低频率状态上报WebSocket好一般断线重连靠自研中中浏览器实时推送自研TCP长连接好取决于实现很高中游戏、自有协议MQTT好好协议本身为弱网设计低低物联网设备通信、即时消息表格里的“实现成本低”初看可能觉得不太直观。其实MQTT对后端开发者来说最大的友好之处在于发布、订阅、QoS这些概念都已经被协议层实现好了我们只需要关心业务消息本身不用去处理粘包拆包、心跳维护、断线重连这些底层难题。1.2 MQTT到底解决了哪三类问题第一类是弱网环境下的可靠性。MQTT基于TCP但它在协议层面设计了重连机制、会话延续和消息QoS等级设备端网络抖一抖重新连上之后还能继续收离线期间的消息这是HTTP轮询很难做到的。第二类是消息过滤与分发。设备端按主题订阅服务端往固定主题发布消息Broker会负责把消息精准推给订阅者。发布者不需要知道谁在订阅订阅者也不需要知道消息从哪里来这对系统解耦帮助很大。第三类是低功耗场景。MQTT的连接保持机制很轻量心跳报文很小设备不需要为了“等一下可能有一条消息”而反复建立连接。我做过的几个设备项目里同样频率的心跳上报用MQTT比用HTTP轮询省电明显。1.3 适合哪类读者学习路径怎么走这篇文章适合已经会用SpringBoot写普通接口、但对消息协议接触不多的后端同学。建议你先别急着看代码先把“发布者-代理-订阅者”这个三角关系在脑子里过一遍服务端既可以作为订阅者接收设备上报也可以作为发布者向设备下发指令。理解了这个模型后面所有代码都是在给这个模型落地。2. 环境与依赖版本搭配决定后续是否顺利SpringBoot整合MQTT我踩过的最早一个坑就是版本搭配。网上能找到的教程五花八门有的基于SpringBoot 1.x有的基于2.x还有的推荐直接引入Spring Integration MQTT模块。如果你照着一个老教程的依赖坐标硬抄启动时大概率会碰到ClassNotFound或者Bean冲突。2.1 版本选型我最终采用的组合我本地和测试环境长期使用的组合是SpringBoot 2.7.x JDK 8/11 Java MQTT客户端库。为什么不用SpringBoot 3.x不是不能用而是3.x基于Jakarta命名空间一些老依赖需要同步升级很多教程和网上的片段还没跟上新手容易把自己绕晕。另外我推荐直接使用Java生态里最常用的MQTT客户端库而不是Spring Integration的封装模块。Spring Integration MQTT确实能提供MessageHandler之类的抽象但它把很多细节藏起来了出了问题不好排查。自封装客户端库虽然多写几个类但每个连接参数都在自己手里线上查问题的时候会舒服很多。2.2 Maven依赖引入与说明在pom.xml里加入下面的依赖dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependency这里用到的版本是目前社区里最稳定、兼容性最好的Java版MQTT客户端库之一。如果你所在团队的私服没有缓存这个版本可以先让Maven重新拉取一次或者临时换成1.2.2功能差别不大但建议线上统一版本避免后续维护混乱。依赖其实就这一个核心的。SpringBoot自带的spring-boot-starter本身不需要额外配置Jackson之类的也不需要因为MQTT消息体默认就是字节数组我们自己决定怎么反序列化。2.3 本地快速搭一个测试用的Broker整合MQTT不能没有一个消息代理我建议你本地用Docker起一个轻量级的MQTT服务端镜像总共就一条命令的事情。docker run -d -p 1883:1883 --name local-mqtt eclipse-mosquitto:2这个容器启动后本地1883端口就是一个完整的MQTT Broker默认是匿名访问。我用这种感觉无压力的环境把代码跑通之后再丢到测试服务器的正式代理上效率高很多。如果你的机器没有Docker也可以安装一个开源的PC端MQTT代理程序Windows和Linux都有对应版本原理一样。启动Broker之后可以用命令行客户端做连通性测试也可以在后面章节的代码写好之后直接用Java客户端连上来验证。注意本地测试Broker的默认配置允许匿名连接但到了正式环境一定要配置账号密码这一点我会在后面配置章节详细说。3. application.yml配置逐项拆解每个字段都是一个坑以前看一些教程配置文件直接贴一坨注释都不写几个看得人一头雾水。我自己把配置拆开逐项解释清楚后才发现很多“连不上”“收不到消息”的问题根源就在配置字段的语义没搞清楚。3.1 一份完整可用的配置模板我习惯把这些参数统一放在以mqtt开头的自定义配置下方便统一管理和加密处理。下面是经过线上验证的一份配置模板可以直接改造使用spring: application: name: mqtt-service mqtt: broker-url: tcp://127.0.0.1:1883 client-id: service-backend-01 username: device-user password: device-pass clean-session: false keep-alive-interval: 60 connection-timeout: 30 auto-reconnect: true topics: - name: device//report qos: 1 - name: gateway/alarm qos: 2这里把topics设计成列表是因为实际项目中通常要同时订阅多个主题而且不同主题的QoS等级可能还不一样。在配置里集中维护比散落在代码里好改得多。3.2 关键字段含义和推荐值配置字段含义推荐值注意事项broker-urlBroker地址tcp://ip:1883SSL场景用ssl://并配证书client-id客户端唯一标识服务名序号同一Broker下必须全局唯一username连接账号Broker中创建的用户匿名测试环境可不填password连接密码强密码加密存储勿明文硬编码clean-session会话清理标志false影响离线消息是否保留keep-alive-interval心跳间隔秒数60需小于网关空闲断开时间connection-timeout连接超时秒数30弱网可适当调大auto-reconnect是否自动重连true正式环境建议加手动重试策略topics订阅主题集合按业务设计支持通配符和#3.3 clientId和cleanSession的关系为什么收不到离线消息这一节是重点几乎每个刚接触MQTT的人都会在这里卡住。我先说结论如果你把clean-session设成true客户端断开后Broker会立即清掉这个客户端的会话信息和所有未消费消息那么离线期间发给它的消息等它重新连上来的时候已经彻底消失了表现就是“消息静悄悄丢了”。反过来把clean-session设成falseBroker会保留客户端的订阅关系和离线消息等客户端重新连接并且用同一个clientId登录时离线期间积压的消息会继续下发。理解了这一点你就不难理解为什么clientId必须保持稳定且唯一。同一个服务实例每次重启连上Broker时如果你换个新的clientId那和之前的订阅关系完全断开了离线消息自然收不到如果两个实例用了同一个clientId则会造成后者把前者踢下线谁都没法稳定工作。所以我在实际配置中会把clientId设计成类似service-name-节点序号的形式每个节点一个固定值重启不换节点之间不重复。设备和后台服务都是这个规则。4. 核心代码连接管理、订阅、回调与发布的完整实现配置到位后代码就是整个集成的骨架。我不喜欢把一堆逻辑堆在一个类里而是分成配置类、连接管理器、订阅回调、发布服务几个部分每个部分只干一件事出问题的时候定位非常快。4.1 配置类把YAML变成强类型对象SpringBoot的配置绑定是最基础的能力先写一个配置类把前面application.yml里的参数映射进来。Data Component ConfigurationProperties(prefix mqtt) public class MqttProperties { private String brokerUrl; private String clientId; private String username; private String password; private boolean cleanSession true; private int keepAliveInterval 60; private int connectionTimeout 30; private boolean autoReconnect true; private ListTopicConfig topics new ArrayList(); Data public static class TopicConfig { private String name; private int qos 1; } }ConfigurationProperties的好处是类型安全还能在启动时就校验字段。比如brokerUrl漏配的时候SpringBoot会直接报错总比运行时连不上再去翻日志强。4.2 连接管理器启动即连接对外提供客户端实例连接管理的核心是把MqttClient的生命周期管理好包括初始化连接、获取当前连接状态、重连触发等。我写成一个MqttConnectionManager由Spring容器管理单例。Component public class MqttConnectionManager { private static final Logger log LoggerFactory.getLogger(MqttConnectionManager.class); private final MqttProperties properties; private MqttClient client; private MqttCallback callback; public MqttConnectionManager(MqttProperties properties) { this.properties properties; } public void connect() throws MqttException { if (client null) { synchronized (this) { if (client null) { client new MqttClient(properties.getBrokerUrl(), properties.getClientId(), new MemoryPersistence()); } } } if (!client.isConnected()) { MqttConnectOptions options buildOptions(); client.setCallback(callback); client.connect(options); log.info(MQTT连接成功clientId{}, properties.getClientId()); } } private MqttConnectOptions buildOptions() { MqttConnectOptions options new MqttConnectOptions(); options.setCleanSession(properties.isCleanSession()); options.setKeepAliveInterval(properties.getKeepAliveInterval()); options.setConnectionTimeout(properties.getConnectionTimeout()); options.setAutomaticReconnect(properties.isAutoReconnect()); if (StringUtils.hasText(properties.getUsername())) { options.setUserName(properties.getUsername()); options.setPassword(properties.getPassword().toCharArray()); } return options; } public MqttClient getClient() { return client; } public boolean isConnected() { return client ! null client.isConnected(); } }这里有几个容易被忽略的细节第一new MqttClient的时候要指定MemoryPersistence否则有些环境会持续写小文件时间久了会出现奇怪的磁盘占用第二setAutomaticReconnect(true)只对客户端断开后的自动重连有效如果connect()本身抛异常该选项并不会无限重试需要我们自己配合重试策略第三client.setCallback必须在connect之前设置否则连接建立后一旦有消息到达回调还没注册消息就丢了。4.3 订阅管理连接成功后注册主题连接成功不代表能收到消息还要主动订阅主题。我把订阅动作抽成一个独立方法既可以在启动阶段调用也可以在动态新增主题时复用。public void subscribeTopics() throws MqttException { if (client null || !client.isConnected()) { throw new IllegalStateException(MQTT客户端未连接无法订阅); } for (MqttProperties.TopicConfig topic : properties.getTopics()) { client.subscribe(topic.getName(), topic.getQos()); log.info(订阅主题成功topic{}qos{}, topic.getName(), topic.getQos()); } }这里需要特别说明QoS的选择。业务上都希望消息可靠所以很多人统一设成QoS2这其实是个误区。QoS2虽然最可靠但吞吐量最低Broker和客户端都要多几轮确认报文。我的经验是普通设备上报用QoS1就够告警、命令这类绝对不能丢的消息才用QoS2而实时性要求高、丢了也无所谓的遥测数据可以用QoS0。订阅之后客户端收到消息时会进入回调方法这是下一步的核心。4.4 消息回调解析、处理、容错都要做全回调是MQTT集成的灵魂。我见过太多项目在回调里直接写业务逻辑结果一个消息处理阻塞后面的消息全部排队积压设备端表现为“服务没响应”。我的做法是回调里只做基础解析和线程池投递真正的业务处理丢给独立线程池。Component public class MqttMessageCallback implements MqttCallback { private static final Logger log LoggerFactory.getLogger(MqttMessageCallback.class); private final ExecutorService messageProcessor; public MqttMessageCallback() { this.messageProcessor new ThreadPoolExecutor( 4, 16, 60, TimeUnit.SECONDS, new LinkedBlockingQueue(2000), new ThreadFactoryBuilder().setNameFormat(mqtt-handler-%d).build(), new ThreadPoolExecutor.CallerRunsPolicy()); } Override public void connectionLost(Throwable cause) { log.error(MQTT连接丢失原因{}, cause.getMessage(), cause); } Override public void messageArrived(String topic, MqttMessage message) { byte[] payload message.getPayload(); messageProcessor.execute(() - { try { String content new String(payload, StandardCharsets.UTF_8); log.info(收到消息topic{}content{}, topic, content); handleMessage(topic, content); } catch (Exception e) { log.error(处理MQTT消息异常topic{}, topic, e); } }); } Override public void deliveryComplete(IMqttDeliveryToken token) { // 发布确认事件可在此记录消息已到达Broker } private void handleMessage(String topic, String content) { // 业务方法根据topic路由到不同处理器 } }CallerRunsPolicy这里是个关键细节。当线程池满时新任务会在调用者线程也就是MQTT客户端回调线程里执行虽然会阻塞后续消息接收但至少不会丢消息。对于物联网场景宁可处理慢一点也不能让消息凭空消失。4.5 发布工具类从服务端向设备下发指令服务端下发指令本质上就是“向某个主题发布消息”。我封装了一个MqttPublisher业务方只需要调用publish(topic, payload)即可。Component public class MqttPublisher { private static final Logger log LoggerFactory.getLogger(MqttPublisher.class); private final MqttConnectionManager connectionManager; public MqttPublisher(MqttConnectionManager connectionManager) { this.connectionManager connectionManager; } public void publish(String topic, String payload, int qos) throws MqttException { ensureConnected(); MqttMessage message new MqttMessage(payload.getBytes(StandardCharsets.UTF_8)); message.setQos(qos); message.setRetained(false); connectionManager.getClient().publish(topic, message); log.info(消息发布成功topic{}qos{}payload{}, topic, qos, payload); } private void ensureConnected() throws MqttException { MqttClient client connectionManager.getClient(); if (client null || !client.isConnected()) { connectionManager.connect(); } } }我在实际项目中还会额外加一个带重试的版本publishWithRetry方法里发布失败时最多重试三次间隔1秒、2秒、4秒。为什么要重试因为MQTT发布时如果连接意外断开client.publish会抛MqttException而这时消息并没有真正到达Broker直接吞掉异常会导致指令丢失这对设备控制类业务是不能接受的。4.6 启动编排先建Bean再连接再订阅最后需要一个入口把这些步骤串起来。我用SmartLifecycle而不是PostConstruct因为PostConstruct执行时机在Spring容器完成依赖注入之后、但无法准确控制多个Bean之间的先后顺序。如果连接管理类的Bean还没有初始化完订阅动作可能拿到空客户端。Component public class MqttLifecycle implements SmartLifecycle { private volatile boolean running false; private final MqttConnectionManager connectionManager; public MqttLifecycle(MqttConnectionManager connectionManager) { this.connectionManager connectionManager; } Override public void start() { try { connectionManager.connect(); // connect成功后注册回调并订阅主题 MqttClient client connectionManager.getClient(); client.setCallback(new MqttMessageCallback()); connectionManager.subscribeTopics(); running true; } catch (MqttException e) { log.error(MQTT启动失败, e); } } Override public void stop() { try { if (connectionManager.getClient() ! null) { connectionManager.getClient().disconnect(); connectionManager.getClient().close(); } } catch (MqttException e) { log.error(MQTT关闭异常, e); } running false; } Override public boolean isRunning() { return running; } Override public int getPhase() { return Integer.MAX_VALUE; } }getPhase返回Integer.MAX_VALUE是让这个Lifecycle在应用启动的最后阶段再执行确保依赖的其他Bean都已经准备就绪。如果你觉得这块比较绕直接用ApplicationRunner也可以效果类似都是为了避免Bean初始化顺序踩坑。5. 亲测有效的排坑记录五个典型事故的完整排查链路这部分是我觉得这篇文章最值钱的地方。网上能找到的整合教程不少但很少有人把“现象是什么、怎么排查、根因在哪、怎么解决”完整写出来。下面五个坑都是我实际工作中碰到过的每一个都能对应到具体场景。5.1 clientId互踢两个实例轮流掉线日志里全是“Lost connection”现象服务部署了两个节点启动后过不了多久节点A的日志出现连接丢失紧接着节点B也出现连接丢失两个节点交替着上线、下线整体服务不可用。排查过程我先看Broker日志发现同一个clientId多次注册连接再看应用配置两个节点的配置都是从配置中心拉取的里面写的clientId是同一个固定值。到这里根因已经清楚了。根因MQTT协议要求每个客户端连接使用唯一的clientId。当新连接使用同一个clientId建立时Broker会主动断开旧连接。两个节点互相抢占形成互踢。解决修改clientId生成规则在配置中心为每个节点维护不同的clientId或者启动时在固定clientId后面拼接一个节点标识。我用的是服务名-节点序号方案节点序号通过环境变量注入简单可靠。5.2 订阅主题收不到消息cleanSession true加重启时机问题现象服务A订阅了device//report主题测试时用客户端工具往该主题发消息服务A收不到但另一个订阅同主题的服务B能收到。排查过程我先确认服务A确实订阅成功启动日志有“订阅主题成功”再确认发布用的主题名和订阅主题匹配。随后我检查了服务A的clean-session配置发现是true而且服务A在测试前曾经重启过几次。关键点来了如果服务A重启前已经订阅了消息但重启时cleanSession为trueBroker会清除旧会话服务A重启后再订阅只能收到之后发布的消息。发布动作发生在服务A重启完全完成之前的那几秒消息自然就丢了。根因测试时发送消息的时机早于客户端重新订阅完成的时机加上cleanSessiontrue不会保留离线消息造成“配置都对但消息没收到”的错觉。解决对需要可靠接收的主题把cleanSession改为false同时确保业务消息是在订阅方法返回之后再发布。排查这类问题还有个技巧用两个客户端一个专门订阅一个专门发布先把发布订阅链路本身验证通再来看业务逻辑。5.3 QoS不一致导致消息丢失发布端QoS1订阅端QoS0现象设备端上报数据用QoS1服务端订阅主题时写的也是QoS1但偶发丢消息统计下来丢掉的大多发生在Broker重启或网络抖动后的短时间内。排查过程我第一次查的时候只看了发布端代码确认是QoS1但忽略了订阅时的QoS实际生效规则。MQTT协议里最终投递的QoS等级由发布端和订阅端共同决定两者中较低的那个生效。如果订阅时请求的是QoS0即使发布端用QoS1发布Broker投递给该订阅者时仍然会降级为QoS0消息没有确认机制网络抖动时就可能丢。根因订阅端配置的QoS低于发布端投递等级被压低。解决把订阅端的QoS改到和发布端一致。如果业务上确实允许丢那就明确接受QoS0如果丢不起两端都要配置QoS1或QoS2并且要在代码注释里写清楚防止后续维护的人改乱。5.4 keepAlive与防火墙冲突连接频繁断开又频繁重连现象服务上线后日志显示连接每隔一两分钟就断一次断开后能自动重连但重连后没多久又断。整体服务可用但消息接收延迟大时断时续。排查过程我先怀疑网络不稳定ping Broker所在的服务器延迟和丢包都正常。接着看网关和防火墙配置发现网络设备里配置了空闲连接超时时间为90秒。而我们MQTT客户端的keepAlive设置为120秒也就是说在这个间隔内客户端没有发送任何MQTT心跳包防火墙会把连接当作空闲连接断开。根因keep-alive-interval设置得比网络中间设备的空闲超时时间更长心跳还没发连接就被外部切断。解决把keepAlive改为45秒或60秒小于网关空闲超时时间即可。同时要注意MQTT协议的keepAlive只是“期望的心跳间隔”客户端库会自动在空闲达到一定比例时发送PINGREQ因此这个值必须和网络链路里的各种超时时间匹配。5.5 Spring容器初始化顺序启动时订阅总是报“客户端未连接”现象本地代码跟着教程写完启动SpringBoot应用时日志报错MQTT客户端未连接无法订阅。但奇怪的是单独用单元测试调用connect再subscribe却是正常的。排查过程我先看启动日志里的Bean初始化顺序发现订阅逻辑所在的Bean先于连接管理Bean执行。原因是我在最开始用的PostConstruct方法里直接调了subscribeTopics()那时MqttConnectionManager虽然被注入了但它自己还没有完成connect()所以客户端处于未连接状态。根因Spring容器中Bean的初始化顺序没有显式编排PostConstruct无法保证相对顺序。解决改用4.6节里的SmartLifecycle方式或者用ApplicationReadyEvent监听器在应用完全启动后再执行连接和订阅。不要指望靠Order注解解决所有顺序问题生命周期接口才是为这种场景设计的。6. 上线前建议加的优化重连、通配符、遗嘱消息与线程模型核心流程跑通后项目距离生产上线还差几步。我把这些优化称为“生产环境四件套”缺了任何一个系统都能跑但稳定性会差一个档次。6.1 动态订阅与多主题通配符业务一复杂主题往往不是写死在配置文件里的而是运行时才能确定。比如某个新设备接入就要订阅一个针对该设备控制的专属主题。此时可以用client.subscribe动态增加订阅不需要重启服务。主题命名的和#通配符也要熟练使用。代表单层通配#代表多层通配。例如device//report匹配device/001/report而device/#匹配device/001/report和device/001/status都能匹配。我建议业务设计上尽量把主题层级控制在三层以内层级越多通配符匹配规则越复杂维护成本越高。6.2 遗嘱消息与保留消息让“掉线”可以被感知设备会掉线服务也会掉线这是常态。MQTT的遗嘱消息Last Will就是用来通知其他订阅者“某个客户端掉线了”。在客户端连接时设置遗嘱Broker会在连接异常断开时替该客户端向指定主题发布一条遗嘱消息。MqttConnectOptions options new MqttConnectOptions(); options.setWill(device/status/offline, {\clientId\:\service-backend-01\}.getBytes(StandardCharsets.UTF_8), 1, false);我在设备状态管理项目里的做法是每个设备上线时发布一条device/status/online的保留消息同时连接设置遗嘱主题为device/status/offline。保留消息意味着新订阅者连上来就能立刻拿到最新的状态不必等设备下次上报。这样后台可以实时维护一张设备在线状态表。6.3 回调线程池永远不要让MQTT的回调线程卡在业务处理上读者可以做一个实验在messageArrived里直接加个Thread.sleep(5000)然后连续给Broker发送多条消息观察客户端接收情况。你会发现后面的消息全部在排队实时性几乎归零。这只是单线程阻塞的演示但现实中调用外部服务超时、数据库慢查询、分布式锁等待等问题都会让回调线程卡住。生产环境我一般把核心线程数设定为设备连接数的十分之一到五分之一最大线程数不超过一个合理上限队列容量要给足。线程池拒绝策略我推荐CallerRunsPolicy宁可阻塞回调不要再丢消息。另外要给线程池里的任务加超时控制比如用Future配合get(timeout)防止个别慢任务长期占用线程。6.4 连接状态监控与消息幂等消费消息没有丢失不代表没有重复。MQTT的QoS1和QoS2在特殊情况下都可能导致重复投递。最典型的是网络抖动时客户端收到消息后还没来得及确认连接就断了重连后Broker会认为消息未投递成功再次下发给客户端。所以集成MQTT的消费端必须考虑幂等。我在处理设备消息时会从消息体中提取一个全局唯一的消息ID在Redis里做去重例如SETNX mqtt:msg:设备ID:消息ID 1 EX 600如果设置成功才执行业务逻辑否则直接跳过。连接状态监控方面可以在connectionLost回调里上报指标也可以在定时任务里检查isConnected()状态。如果项目接入了监控系统建议把MQTT的连接状态、消息积压量、线程池活跃度都暴露出去超过阈值直接告警不然线上出了问题很难快速发现是MQTT链路的问题。最后再分享一个小技巧每次我搭建MQTT测试环境都会在本地同时开一个带图形界面的MQTT客户端工具一边用Java代码发布消息一边用图形工具订阅同一个主题这样能直观看到消息流转是否正常。遇到Java客户端收不到消息的诡异问题先别急着改代码用图形工具代替Java端做订阅如果图形工具能收到问题大概率在Java代码配置上如果图形工具也收不到那就回头查Broker权限、主题命名和QoS设置。我在本地调试时还有一个习惯把日志级别调整到DEBUG重点观察connectionLost和messageArrived两个方法。MQTT链路的问题九成都能从这两个方法的关键日志里找到蛛丝马迹。搞懂上面这些配置和坑之后SpringBoot整合MQTT就不再是什么玄学问题了按部就班地搭你也能一次跑通。
