接手一个若依前后端分离版项目第一件事往往不是写业务而是先想清楚“外部设备的数据怎么进来”。最近在做充电桩物联网平台时我需要把设备上报的状态、电量、告警信息实时接入若依后台选来选去还是走MQTT这条线最稳。这篇就把完整的落地过程写出来从协议原理到若依后端集成再到MQTTX这个测试工具怎么用一步步走一遍。如果你正打算在若依框架里接入物联网设备或者只是因为项目需要搞懂MQTT是怎么和Spring Boot配合的照着这篇文章操作就能跑通。1. 为什么在若依里集成MQTT场景选型与方案权衡1.1 若依MQTT能解决什么真实业务问题若依本身是个非常成熟的管理后台框架用户、角色、菜单、权限、操作日志这些基础能力都有做企业内部系统效率很高。但有个典型盲区它天生不是为实时数据设计的。你用若依做设备管理平台、充电桩监控、环境监测系统这类带硬件设备接入的项目时会发现设备上报数据、平台下发指令、实时告警推送这些场景用传统的HTTP请求轮询来做体验和性能都非常差。MQTT恰好补上这块短板。它是专门为物联网和弱网环境设计的轻量级消息传输协议基于发布/订阅模型一条消息从设备端发出Broker转发订阅了对应主题的服务端立刻收到。和HTTP最大的区别是HTTP是客户端主动拉取MQTT是Broker主动推送服务端不需要轮询延迟能控制在毫秒级。我这次做充电桩项目桩端通过4G模块连接MQTT Broker定时上报电压、电流、SOC、充电状态等数据。若依后台负责设备管理、用户管理、订单计费这些核心业务底层通过集成MQTT客户端接收设备消息把实时状态写入数据库并推送到前端页面。这样若依不再只是一个纯管理后台而是变成了一个真正能和硬件设备联动的物联网平台底座。1.2 三种集成方案怎么选Paho、Spring Integration MQTT、自研长连接在若依这种Spring Boot项目里集成MQTT社区里有几种主流做法我对比之后选了最适合的一套。第一类是直接用Eclipse Paho Java Client。这是Eclipse基金会出的MQTT客户端库独立于Spring生态用法非常直观创建MqttClient注册回调连接Broker然后订阅主题、处理消息。它的优点是轻、灵活、可控性强断线重连、遗嘱消息这些都能在代码里精确控制适合想要完全掌握连接生命周期的场景。第二类是Spring Integration MQTT。这是Spring官方提供的一套集成方案把MQTT封装成了Spring Integration的消息通道MessageChannel通过注解和配置文件来声明消息流。优点是和Spring生态贴合度高代码量少但调试时比较绕很多细节被框架封装了出了问题不好定位。第三类是自研长连接比如基于Netty手写一个MQTT协议解析器。这种方案技术含量高灵活性最强但代价是要处理协议细节、心跳机制、粘包拆包、认证鉴权等一系列问题开发周期长对大多数项目来说属于过度设计。我的建议是项目工期紧、以功能交付为主直接用Paho如果团队对Spring Integration很熟、且项目里已有大量Spring Integration组件可以选第二套除非你是做中间件产品的厂商否则没必要自研。实际集成时还有一个细节需要注意Paho的artifactId有两个版本一个是org.eclipse.paho:org.eclipse.paho.client.mqttv3这是官方原生Java客户端另一个是org.eclipse.paho:org.eclipse.paho.mqttv5.client对应MQTT 5.0协议。目前大部分Broker和硬件设备还在用3.1.1协议兼容性最好所以下面教程里我以mqttv3为例。如果设备或Broker明确支持MQTT 5.0特性比如消息过期、主题别名这类的再考虑升级到v5客户端。1.3 整体架构设计后端订阅、回调分发、WebSocket推送到前端整条数据链路要从前到后打通不能后端收到消息就算完事页面上的实时状态才是业务人员真正看到的东西。所以我设计了这样一套架构也推荐你按这个思路来硬件设备/模拟端通过MQTT协议连接Broker向指定主题发布消息比如充电桩向device/{deviceId}/status发布设备状态数据。Broker消息代理我这里用的是EMQX作用相当于消息中转站负责接收设备消息再分发给所有订阅了这个主题的客户端。若依后端集成Paho MQTT客户端启动后连接Broker并订阅主题通过回调方法接收消息。回调里做数据解析、落库同时把消息通过WebSocket推送到前端。若依前端Vue通过WebSocket接收实时消息渲染成图表、表格或弹窗告警。这套架构的好处是思路清晰、每层职责单一。MQTT负责设备端到服务端的实时传输WebSocket负责服务端到浏览器的实时推送两者互补没有谁替代谁的问题。数据落库用的是若依框架内置的MyBatis和Service层权限控制也能直接复用若依的体系。2. MQTT协议要点与MQTTX工具准备2.1 MQTT核心概念十分钟速通开始写代码之前有几个概念必须先理解否则后面配置参数时容易一头雾水。Broker是整个MQTT体系的心脏所有消息都经过它转发。它不产生消息只负责接收发布者发来的消息、匹配订阅条件、推送给订阅者。常见的Broker软件有EMQX、Mosquitto、HiveMQ等用户根据系统规模选就行。**Topic主题**是消息的分类标识采用斜杠层级结构比如device/001/status。订阅方用通配符来匹配一类主题device//status匹配任意设备ID的status主题device/#匹配device下所有层级。这个设计非常像文件系统目录理解起来零门槛。**QoS服务质量**决定消息投递的可靠性级别有三个档位QoS 0最多一次发出去就不管可能丢消息适合普通遥测数据。QoS 1至少一次保证消息到达但可能重复适合大部分业务场景。QoS 2恰好一次通过两阶段握手确保不重不漏性能开销最大适合计费、指令下发等对准确性要求极高的场景。**遗嘱消息Last Will**是一个很实用的机制。客户端连接Broker时可以设置一条遗嘱消息如果客户端非正常断开比如断网、断电Broker会代替客户端把遗嘱消息发布到指定主题。利用这个特性可以感知设备异常离线对物联网场景很重要。2.2 本地搭建Broker基于EMQX的安装与配置为了本地开发和测试一条消息发出去需要有Broker接收和转发。我选了EMQX轻量、开源、管理界面好用而且官方支持Docker一键部署。如果你本地装了Docker可以直接跑docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx:5.8.0端口说明1883MQTT普通TCP端口本地开发主要走这个。8883MQTT SSL端口生产环境走这个。8083MQTT over WebSocket端口浏览器端MQTT连接用。18083EMQX Dashboard管理界面端口浏览器打开http://localhost:18083就能登录默认账号admin/public。启动后浏览器访问管理界面可以在里面看到连接数、消息数、订阅情况还可以在线发测试消息非常方便。生产环境部署时建议用ACL控制用户权限并为不同的设备分配独立的用户名和主题访问权限。2.3 MQTTX连接测试超好用的MQTT客户端工具箱没有MQTTX之前测试MQTT消息通常用命令行工具或写测试代码非常麻烦。MQTTX是EMQ官方出的跨平台桌面客户端界面长得像聊天软件左边是连接列表中间是消息收发记录右侧是操作面板上手几乎没有学习成本。MQTTX的下载与安装可以直接去GitHub Releases页面下载对应操作系统的安装包支持Windows、macOS、Linux另外还有Android和iOS手机版。你搜索“MQTTX下载”就能找到官方地址。装好之后打开界面非常简洁新手不会有任何迷茫感。新建一个连接的配置项如下Name连接名称随意填写比如“本地测试”。HostBroker地址格式是mqtt://127.0.0.1:1883。注意MQTTX要求带协议前缀本地没加密就用mqtt://TLS加密就填mqtts://。Username/Password如果Broker开了认证就填写。本地EMQX默认允许匿名可以留空。Client ID客户端唯一标识默认会自动生成建议保持唯一。如果两个客户端用同一个ClientID连接同一个Broker前者会被强制踢下线这个坑后面还会讲到。Clean Session是否清除会话测试时勾选即可。点击“连接”按钮中间面板显示“Connected”就说明连接成功了。然后订阅一个主题比如device/test再往这个主题发送一条JSON消息比如{temperature: 25.5}立刻就能在订阅端看到这条消息。整个流程不到一分钟非常直观。MQTTX还有一些高级功能也很实用。比如它可以在一个连接里同时订阅多个主题可以给服务端发送遗嘱消息测试离线感知可以复制一条消息重新编辑还有脚本功能能自定义消息生成规则模拟多设备上报。这些在调试阶段能省下大量时间。3. 若依后端集成MQTT完整实操3.1 引入依赖与自定义配置项我在若依后端项目的pom.xml里添加Paho客户端依赖。dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependency然后在application.yml中添加自定义的MQTT相关配置。因为若依框架本身有多环境配置application-dev.yml、application-prod.yml我把MQTT配置放在公共配置里方便不同环境覆盖。mqtt: host: tcp://127.0.0.1:1883 client-id: ruoyi-server-001 username: admin password: public topic: device/# qos: 1 completion-timeout: 3000 keep-alive-interval: 60参数含义说明hostBroker地址格式tcp://ip:port。client-id客户端唯一标识服务端实例多部署时要注意唯一否则会互相踢下线。username/password连接Broker的认证信息。topic服务端启动后默认订阅的主题可以用通配符。qos订阅的默认服务质量等级。keep-alive-interval心跳间隔单位秒客户端会在这个时间周期内发送PINGREQBroker连续一段时间没收到心跳就判定连接断开。3.2 客户端初始化连接参数、自动重连、主题订阅接下来创建配置类在Spring Boot启动时初始化MQTT客户端并建立连接。这一段是核心照着写就能跑通。package com.ruoyi.framework.mqtt; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; Data Component ConfigurationProperties(prefix mqtt) public class MqttProperties { private String host; private String clientId; private String username; private String password; private String topic; private int qos; private int completionTimeout; private int keepAliveInterval; }package com.ruoyi.framework.mqtt; import lombok.extern.slf4j.Slf4j; import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import javax.annotation.Resource; Slf4j Configuration public class MqttConfig { Resource private MqttProperties mqttProperties; Resource private MqttMessageCallback mqttMessageCallback; Bean public MqttClient mqttClient() throws Exception { MemoryPersistence persistence new MemoryPersistence(); MqttClient client new MqttClient( mqttProperties.getHost(), mqttProperties.getClientId(), persistence ); MqttConnectOptions options new MqttConnectOptions(); options.setUserName(mqttProperties.getUsername()); options.setPassword(mqttProperties.getPassword().toCharArray()); options.setCleanSession(true); options.setConnectionTimeout(mqttProperties.getCompletionTimeout()); options.setKeepAliveInterval(mqttProperties.getKeepAliveInterval()); // 开启自动重连 options.setAutomaticReconnect(true); // 设置遗嘱消息 options.setWill(device/server/status, offline.getBytes(), 1, false); client.setCallback(mqttMessageCallback); client.connect(options); if (client.isConnected()) { client.subscribe(mqttProperties.getTopic(), mqttProperties.getQos()); log.info(MQTT连接成功已订阅主题{}, mqttProperties.getTopic()); } return client; } }这里有三个点值得展开说。第一为什么用MemoryPersistencePaho客户端支持将消息持久化到磁盘保证客户端重启后未发送完的消息还能继续处理。但若依通常部署在标准化服务器环境里磁盘路径配置麻烦内存持久化简单可靠配合QoS 1已经能满足绝大多数业务。如果真有严格的消息补偿需求建议在业务层做幂等处理而不是依赖客户端持久化。第二为什么在订阅时用通配符device/#我按业务规划把所有设备相关消息都收敛到device这个父级主题下服务端一次订阅就能覆盖所有设备状态。通配符是很方便但要注意安全性和消息量如果所有消息都混在一个大主题下后端处理的压力会很大。所以规划主题时要有层级设计比如device/{deviceId}/status、device/{deviceId}/cmd/reply代码里再按topic段解析出设备ID。第三遗嘱消息的用途。我设了一条device/server/status的遗嘱消息内容是offline。如果若依服务端进程崩溃Broker会立刻把这个遗嘱消息发出去订阅这个主题的运维系统就能第一时间感知到服务端离线了。这是很实用的高可用设计平时不会有感知故障时能省下大量排查时间。3.3 消息回调处理解析数据、入库、推送消息回调是整个集成的核心枢纽。设备上报的所有数据都会汇聚到这里所以处理的逻辑要设计得清晰、可扩展。我的做法是回调里只做消息解析和分发具体的业务处理交给不同的Service。package com.ruoyi.framework.mqtt; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.ruoyi.system.service.IDeviceDataService; import com.ruoyi.system.service.IWebSocketService; import lombok.extern.slf4j.Slf4j; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttCallbackExtended; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.springframework.stereotype.Component; import javax.annotation.Resource; Slf4j Component public class MqttMessageCallback implements MqttCallbackExtended { Resource private IDeviceDataService deviceDataService; Resource private IWebSocketService webSocketService; Override public void connectComplete(boolean reconnect, String serverURI) { log.info(MQTT连接完成重连标志{}, reconnect); // 断线重连成功后需要重新订阅主题 // 实际项目中可以在这里重新订阅 } Override public void connectionLost(Throwable cause) { log.error(MQTT连接丢失, cause); // AutomaticReconnect开启后这里只记录日志重连由框架处理 } Override public void messageArrived(String topic, MqttMessage message) { String payload new String(message.getPayload()); log.info(收到消息主题{}内容{}, topic, payload); try { JSONObject json JSON.parseObject(payload); // 从主题中解析设备ID例如 device/CH001/status String[] topicParts topic.split(/); String deviceId topicParts.length 2 ? topicParts[1] : null; // 业务分发 if (topic.endsWith(/status)) { deviceDataService.saveDeviceStatus(deviceId, json); } else if (topic.endsWith(/alarm)) { deviceDataService.saveAlarm(deviceId, json); } // 推送给前端 webSocketService.sendMessage(topic, payload); } catch (Exception e) { log.error(MQTT消息处理失败, e); } } Override public void deliveryComplete(IMqttDeliveryToken token) { // 消息发布完成回调一般用于指令下发确认 } }消息处理里最重要的原则是回调方法里不要做耗时操作。因为MQTT的messageArrived是单线程回调如果这里去查数据库、调用外部接口、写日志文件消息处理的吞吐量会严重下降。我实际调试时发现当设备量大之后处理不过来会导致消息积压、延迟增大。正确做法是收到消息后立即放入线程池异步处理Resource private ThreadPoolTaskExecutor mqttTaskExecutor; Override public void messageArrived(String topic, MqttMessage message) { mqttTaskExecutor.execute(() - processMessage(topic, message.getPayload())); }这样消息到达后立即返回业务处理放到独立线程池中执行互不阻塞。线程池的配置可以在若依框架里直接复用现有的异步任务线程池也可以单独创建一个核心线程数和队列大小根据消息量来定我这次按每秒20条消息的业务量配了8个核心线程、队列容量200。3.4 动态主题订阅在若依后台维护设备主题有一种常见需求设备上线后后台管理系统需要给这台设备下发指令就涉及向指定主题发布消息。还有一种更细的场景不同设备类型只需要订阅对应的主题而不是所有设备。这时动态订阅机制就显得很重要。在若依框架里我做了这样一个设计设备表新增一个topic字段设备接入时后台新增设备并记录该设备的主题。启动时先订阅默认主题device/#接收所有消息同时通过若依的定时任务或者设备上线的接口调用订阅管理服务动态增加或减少订阅。package com.ruoyi.framework.mqtt; import lombok.extern.slf4j.Slf4j; import org.eclipse.paho.client.mqttv3.MqttClient; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; Slf4j Service public class MqttSubscribeService { Resource private MqttClient mqttClient; /** * 记录已订阅的主题避免重复订阅 */ private final SetString subscribedTopics ConcurrentHashMap.newKeySet(); public void subscribe(String topic, int qos) { if (subscribedTopics.contains(topic)) { log.info(主题 {} 已订阅过跳过, topic); return; } try { mqttClient.subscribe(topic, qos); subscribedTopics.add(topic); log.info(动态订阅主题成功{}, topic); } catch (Exception e) { log.error(动态订阅主题失败{}, topic, e); } } public void unsubscribe(String topic) { try { mqttClient.unsubscribe(topic); subscribedTopics.remove(topic); log.info(取消订阅主题成功{}, topic); } catch (Exception e) { log.error(取消订阅主题失败{}, topic, e); } } }这里用ConcurrentHashMap的keySet作为线程安全的订阅主题集合是为了避免并发场景下重复订阅的问题。在若依的Service层调用这个服务时你需要在业务Service里注入MqttSubscribeService在设备新增、删除的接口里加上对应的订阅和取消逻辑。但有一个容易忽略的点断线重连后Paho客户端不会自动恢复之前动态订阅的主题。自动重连机制只恢复连接不恢复会话和订阅。所以我在MqttMessageCallback的connectComplete方法里把MqttSubscribeService里维护的主题集合重新订阅一遍Override public void connectComplete(boolean reconnect, String serverURI) { log.info(MQTT连接完成重连标志{}重新订阅动态主题, reconnect); subscribedTopics.forEach(topic - { try { if (!mqttClient.isConnected()) { return; } // 实际项目中这里需要重新遍历订阅 // mqttClient.subscribe(topic, qos); } catch (Exception e) { log.error(重新订阅失败{}, topic, e); } }); }如果你在前面的阶段没有考虑这个细节生产环境一旦网络抖动触发重连部分设备消息就会静默丢失排查起来非常头疼。4. 前端实时展示Vue3中接入WebSocket4.1 后端提供WebSocket服务若依框架本身没有内置WebSocket支持需要自己加一个WebSocket服务端。好在Spring Boot对WebSocket支持很完善新建一个配置类启用WebSocket再写一个端点类处理连接和消息推送。package com.ruoyi.framework.websocket; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.config.annotation.WebSocketConfigurer; import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; import javax.annotation.Resource; Configuration EnableWebSocket public class WebSocketConfig implements WebSocketConfigurer { Resource private WebSocketServer webSocketServer; Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { registry.addHandler(webSocketServer, /ws/mqtt) .setAllowedOrigins(*); } }WebSocket服务端核心类package com.ruoyi.framework.websocket; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; import java.util.concurrent.ConcurrentHashMap; Slf4j Component public class WebSocketServer extends TextWebSocketHandler { private static final ConcurrentHashMapString, WebSocketSession SESSION_MAP new ConcurrentHashMap(); Override public void afterConnectionEstablished(WebSocketSession session) { SESSION_MAP.put(session.getId(), session); log.info(WebSocket连接建立{}当前连接数{}, session.getId(), SESSION_MAP.size()); } Override protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { // 接收客户端消息通常用于心跳或指令 } Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { SESSION_MAP.remove(session.getId()); log.info(WebSocket连接关闭{}, session.getId()); } public void sendToAll(String topic, String payload) { TextMessage message new TextMessage(payload); SESSION_MAP.forEach((id, session) - { try { if (session.isOpen()) { session.sendMessage(message); } } catch (Exception e) { log.error(WebSocket推送失败{}, id, e); } }); } }然后在之前的MQTT消息回调里把解析好的消息直接调用webSocketServer.sendToAll(topic, payload)推到前端。注意这里做了个简化处理没有区分用户权限。实际项目中建议结合若依的登录用户体系按用户订阅的设备主题做定向推送避免用户A看到用户B的设备数据。4.2 Vue组件中建立连接与消息渲染前端Vue这边比较简单组件挂载时创建WebSocket连接收到消息更新页面数据。export default { name: DeviceMonitor, data() { return { deviceStatusList: [], ws: null } }, created() { this.initWebSocket() }, beforeUnmount() { if (this.ws) { this.ws.close() } }, methods: { initWebSocket() { const protocol location.protocol https: ? wss : ws const wsUrl ${protocol}://${location.host}/ws/mqtt this.ws new WebSocket(wsUrl) this.ws.onopen () { console.log(WebSocket连接成功) } this.ws.onmessage (event) { const data JSON.parse(event.data) this.handleMqttMessage(data) } this.ws.onclose () { console.log(WebSocket连接断开3秒后重连) setTimeout(() { this.initWebSocket() }, 3000) } this.ws.onerror (error) { console.error(WebSocket连接错误, error) } }, handleMqttMessage(data) { // 根据设备ID更新列表 const index this.deviceStatusList.findIndex(item item.deviceId data.deviceId) if (index -1) { this.$set(this.deviceStatusList, index, data) } else { this.deviceStatusList.push(data) } } } }画面上用若依自带的Table组件展示设备列表用Tag组件显示在线状态、Alert组件显示告警再配一个ECharts折线图展示电压电流变化就是一个很完整的设备监控页面。前端这块本身不复杂真正需要注意的还是WebSocket的心跳与断线重连浏览器不会帮你维持连接网络抖动就可能断开所以我在onclose里做了3秒自动重连。不过这里有个坑如果你在MQTT消息回调里给前端推的是原始消息前端每次都要解析主题字符串、判断消息类型逻辑会越写越乱。建议后端在推送前先包装一层比如统一格式化成{topic: device/CH001/status, deviceId: CH001, type: status, data: {...}, timestamp: 1690000000000}前端拿到一个完整自洽的消息对象直接渲染就行。5. 常见问题与排查实录5.1 连接不上Broker怎么排查这个坑几乎每个新手都会踩一遍我把排查路径整理成清单出了问题按顺序对照现象可能原因解决办法connect timed out网络不通或Broker未启动先ping服务器IP再用telnet ip 1883测试端口Connection refusedBroker端口没监听或防火墙拦截检查EMQX是否启动确认1883端口监听Not authorized to connect用户名或密码错误在EMQX Dashboard里新建用户确认配置正确ClientId already in use有另一个客户端用了相同ClientID更换ClientID确保全局唯一MqttException: Unexpected errorbroker版本和客户端不兼容确认broker支持MQTT 3.1.1用mqttv3的兼容性最好排查连接问题有一个技巧先不管若依代码直接用MQTTX试着连接同一个Broker。MQTTX能连上说明Broker和网络没问题问题就出在若依后端的配置上MQTTX也连不上那就先解决Broker的问题。这个二分法能快速缩小范围。5.2 消息收不到/重复收到是怎么回事这类问题表象相似但原因差别很大要结合具体场景来判断。收不到消息先看三处第一Broker管理界面里能看到这个Topic的消息吗看不到说明设备根本没发上来。第二若依后端的订阅主题和设备发布的主题是否匹配device/001/status和device/001/state就差一个字符消息就是收不到。第三QoS等级是否设置合理有些设备端用QoS 0发送Broker到服务端的订阅也用QoS 0任何网络抖动都可能丢消息。对于关键业务尽量在两端都配置QoS 1。重复收到消息最常见的原因是QoS 1的“至少一次”投递语义本身就允许重复。解决方式不是去改QoS而是业务处理时做幂等。我在设备数据上报里加了消息ID字段每条消息携带唯一的messageId数据库里加唯一索引。消息重复到达时插入操作直接报唯一键冲突在代码里捕获这个异常并忽略即可。消息丢失除了网络问题还有一种隐蔽的原因客户端设置了Clean Session为true连接断开期间的离线消息全部丢弃。如果业务对消息完整性要求高需要设置Clean Session为false并开启持久化会话代价是Broker需要为每个客户端缓存未确认的消息内存开销要大一些。5.3 断线重连失效与ClientID冲突Paho客户端的AutomaticReconnect开启后断线后会自动尝试重连但在某些情况下还是不生效。比如网络长时间中断后重新恢复自动重连机制会一直尝试但没有退避策略频繁的重连请求会增加Broker压力。实际项目中建议自己实现重连逻辑在connectionLost回调里用指数退避算法1秒、2秒、4秒...最大60秒重连避免服务端和Broker都被拖垮。另一个经典问题是多实例部署时的ClientID冲突。若依微服务版如果同一服务部署了多个实例所有实例都用同一个ClientID连接同一个Broker后面连接的会顶掉前面的。解决方式是在配置里给ClientID加上实例标识比如ruoyi-server-${spring.cloud.client.ip-address}-${server.port}保证每个实例唯一。如果用若依微服务版还需要注意一个问题多个微服务如果都集成了MQTT客户端同一主题会被多个服务同时消费造成重复处理。这时候要么拆分主题每个服务只订阅自己关心的主题要么引入消息队列多个服务订阅同一个主题但消息只在其中一个服务中处理需要自行设计分布式锁或消费组机制。5.4 若依相关集成注意事项与避坑总结最后整理几个和若依框架结合时的特殊问题这些都是实际项目中容易忽略的。第一若依的权限控制要延伸到MQTT主题。若依框架对HTTP接口有完善的角色权限控制但MQTT是独立通道如果不做任何处理任何拿到Broker账号的人都能订阅设备主题、获取全部设备数据。这里建议做两层控制一是Broker层面的ACL按设备维度限制用户只能订阅自己的主题二是服务端在消息回调里增加校验检查设备ID是否与当前系统中的设备匹配。第二若依定时任务可以和MQTT结合做指令下发。比如运营人员每天凌晨通过后台设置充电策略到点后需要给设备下发指令。常规做法是写一个若依的定时任务到点后调用MqttPublishService向设备主题发布控制消息。Paho的MqttClient里发布消息很简单public void publish(String topic, String payload, int qos, boolean retained) { MqttMessage message new MqttMessage(payload.getBytes()); message.setQos(qos); message.setRetained(retained); mqttClient.publish(topic, message); }第三若依Vue3版本TS报错问题。新版若依Vue3TypeScript项目结构会多一些类型定义集成WebSocket时如果把事件数据直接当any类型处理编译时可能报TS错误。建议定义好接口类型比如MqttMessagePayload使用JSON.parse(JSON.stringify(data))做类型转换时注意类型断言。我在实际开发中就遇到过一个诡异的时间问题类型断言写得不严谨导致字符串当成对象处理页面直接白屏后来统一走接口类型定义加运行时校验才稳定下来。第四不要忘了给若依的系统监控模块加MQTT连接状态的监控。若依自带定时任务和系统监控页面可以在系统监控里增加一个MQTT连接状态指标通过Actuator暴露健康信息再配置告警。这样MQTT连接断开时运维能第一时间收到通知而不是等用户反馈“页面数据不动了”才发现问题。第五数据量上来后要做落库优化。设备状态消息往往频率很高如果每条消息都直接插入数据库数据库压力会非常大。我在这个项目里是先把原始数据存到Redis列表定时任务每隔10秒批量写入MySQL或者用ClickHouse这种时序数据库存海量设备数据。MySQL写不了高频数据这在物联网场景是个常识但很多第一次做的人都会踩这个坑。最后分享一个调试心得集成过程中最花时间的往往不是写代码而是定位“消息到底走到哪一步了”。这是所有消息系统调试的难点MQTT尤其如此——消息在经过设备、Broker、后端、WebSocket、前端五个节点后任何一个环节出错都会表现为“页面没数据”或“数据不对”。我的习惯是先在MQTTX里订阅后端要订阅的主题确认设备此时是否在发消息再确认后端日志里是否打印了messageArrived的收到记录接着确认WebSocket服务端SESSION_MAP里是否有前端连接最后看前端控制台WebSocket有没有收到消息。这一步一步往下排问题一定出现在最后一个正常节点的下游。只要保持这个排查逻辑再复杂的问题也能快速定位。另外再提一句调试MQTT时日志一定要打好。我强烈建议在消息处理的多个关键点打上不同级别的日志收到消息打DEBUG级别并带主题和内容摘要入库成功打DEBUG推送WebSocket打DEBUG异常打ERROR并带上完整异常栈。这样排查问题时日志就是你的王牌debug效率能翻好几倍。希望这篇文章能帮你少踩几个坑照着做完能跑通一个完整的若依MQTT项目然后再根据自己的业务场景去扩展优化。 
长河君 · 2026/9/16 2:00:07 · 阅读 1.2k 若依框架集成MQTT实现设备接入：从原理到实践
接手一个若依前后端分离版项目第一件事往往不是写业务而是先想清楚“外部设备的数据怎么进来”。最近在做充电桩物联网平台时我需要把设备上报的状态、电量、告警信息实时接入若依后台选来选去还是走MQTT这条线最稳。这篇就把完整的落地过程写出来从协议原理到若依后端集成再到MQTTX这个测试工具怎么用一步步走一遍。如果你正打算在若依框架里接入物联网设备或者只是因为项目需要搞懂MQTT是怎么和Spring Boot配合的照着这篇文章操作就能跑通。1. 为什么在若依里集成MQTT场景选型与方案权衡1.1 若依MQTT能解决什么真实业务问题若依本身是个非常成熟的管理后台框架用户、角色、菜单、权限、操作日志这些基础能力都有做企业内部系统效率很高。但有个典型盲区它天生不是为实时数据设计的。你用若依做设备管理平台、充电桩监控、环境监测系统这类带硬件设备接入的项目时会发现设备上报数据、平台下发指令、实时告警推送这些场景用传统的HTTP请求轮询来做体验和性能都非常差。MQTT恰好补上这块短板。它是专门为物联网和弱网环境设计的轻量级消息传输协议基于发布/订阅模型一条消息从设备端发出Broker转发订阅了对应主题的服务端立刻收到。和HTTP最大的区别是HTTP是客户端主动拉取MQTT是Broker主动推送服务端不需要轮询延迟能控制在毫秒级。我这次做充电桩项目桩端通过4G模块连接MQTT Broker定时上报电压、电流、SOC、充电状态等数据。若依后台负责设备管理、用户管理、订单计费这些核心业务底层通过集成MQTT客户端接收设备消息把实时状态写入数据库并推送到前端页面。这样若依不再只是一个纯管理后台而是变成了一个真正能和硬件设备联动的物联网平台底座。1.2 三种集成方案怎么选Paho、Spring Integration MQTT、自研长连接在若依这种Spring Boot项目里集成MQTT社区里有几种主流做法我对比之后选了最适合的一套。第一类是直接用Eclipse Paho Java Client。这是Eclipse基金会出的MQTT客户端库独立于Spring生态用法非常直观创建MqttClient注册回调连接Broker然后订阅主题、处理消息。它的优点是轻、灵活、可控性强断线重连、遗嘱消息这些都能在代码里精确控制适合想要完全掌握连接生命周期的场景。第二类是Spring Integration MQTT。这是Spring官方提供的一套集成方案把MQTT封装成了Spring Integration的消息通道MessageChannel通过注解和配置文件来声明消息流。优点是和Spring生态贴合度高代码量少但调试时比较绕很多细节被框架封装了出了问题不好定位。第三类是自研长连接比如基于Netty手写一个MQTT协议解析器。这种方案技术含量高灵活性最强但代价是要处理协议细节、心跳机制、粘包拆包、认证鉴权等一系列问题开发周期长对大多数项目来说属于过度设计。我的建议是项目工期紧、以功能交付为主直接用Paho如果团队对Spring Integration很熟、且项目里已有大量Spring Integration组件可以选第二套除非你是做中间件产品的厂商否则没必要自研。实际集成时还有一个细节需要注意Paho的artifactId有两个版本一个是org.eclipse.paho:org.eclipse.paho.client.mqttv3这是官方原生Java客户端另一个是org.eclipse.paho:org.eclipse.paho.mqttv5.client对应MQTT 5.0协议。目前大部分Broker和硬件设备还在用3.1.1协议兼容性最好所以下面教程里我以mqttv3为例。如果设备或Broker明确支持MQTT 5.0特性比如消息过期、主题别名这类的再考虑升级到v5客户端。1.3 整体架构设计后端订阅、回调分发、WebSocket推送到前端整条数据链路要从前到后打通不能后端收到消息就算完事页面上的实时状态才是业务人员真正看到的东西。所以我设计了这样一套架构也推荐你按这个思路来硬件设备/模拟端通过MQTT协议连接Broker向指定主题发布消息比如充电桩向device/{deviceId}/status发布设备状态数据。Broker消息代理我这里用的是EMQX作用相当于消息中转站负责接收设备消息再分发给所有订阅了这个主题的客户端。若依后端集成Paho MQTT客户端启动后连接Broker并订阅主题通过回调方法接收消息。回调里做数据解析、落库同时把消息通过WebSocket推送到前端。若依前端Vue通过WebSocket接收实时消息渲染成图表、表格或弹窗告警。这套架构的好处是思路清晰、每层职责单一。MQTT负责设备端到服务端的实时传输WebSocket负责服务端到浏览器的实时推送两者互补没有谁替代谁的问题。数据落库用的是若依框架内置的MyBatis和Service层权限控制也能直接复用若依的体系。2. MQTT协议要点与MQTTX工具准备2.1 MQTT核心概念十分钟速通开始写代码之前有几个概念必须先理解否则后面配置参数时容易一头雾水。Broker是整个MQTT体系的心脏所有消息都经过它转发。它不产生消息只负责接收发布者发来的消息、匹配订阅条件、推送给订阅者。常见的Broker软件有EMQX、Mosquitto、HiveMQ等用户根据系统规模选就行。**Topic主题**是消息的分类标识采用斜杠层级结构比如device/001/status。订阅方用通配符来匹配一类主题device//status匹配任意设备ID的status主题device/#匹配device下所有层级。这个设计非常像文件系统目录理解起来零门槛。**QoS服务质量**决定消息投递的可靠性级别有三个档位QoS 0最多一次发出去就不管可能丢消息适合普通遥测数据。QoS 1至少一次保证消息到达但可能重复适合大部分业务场景。QoS 2恰好一次通过两阶段握手确保不重不漏性能开销最大适合计费、指令下发等对准确性要求极高的场景。**遗嘱消息Last Will**是一个很实用的机制。客户端连接Broker时可以设置一条遗嘱消息如果客户端非正常断开比如断网、断电Broker会代替客户端把遗嘱消息发布到指定主题。利用这个特性可以感知设备异常离线对物联网场景很重要。2.2 本地搭建Broker基于EMQX的安装与配置为了本地开发和测试一条消息发出去需要有Broker接收和转发。我选了EMQX轻量、开源、管理界面好用而且官方支持Docker一键部署。如果你本地装了Docker可以直接跑docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx:5.8.0端口说明1883MQTT普通TCP端口本地开发主要走这个。8883MQTT SSL端口生产环境走这个。8083MQTT over WebSocket端口浏览器端MQTT连接用。18083EMQX Dashboard管理界面端口浏览器打开http://localhost:18083就能登录默认账号admin/public。启动后浏览器访问管理界面可以在里面看到连接数、消息数、订阅情况还可以在线发测试消息非常方便。生产环境部署时建议用ACL控制用户权限并为不同的设备分配独立的用户名和主题访问权限。2.3 MQTTX连接测试超好用的MQTT客户端工具箱没有MQTTX之前测试MQTT消息通常用命令行工具或写测试代码非常麻烦。MQTTX是EMQ官方出的跨平台桌面客户端界面长得像聊天软件左边是连接列表中间是消息收发记录右侧是操作面板上手几乎没有学习成本。MQTTX的下载与安装可以直接去GitHub Releases页面下载对应操作系统的安装包支持Windows、macOS、Linux另外还有Android和iOS手机版。你搜索“MQTTX下载”就能找到官方地址。装好之后打开界面非常简洁新手不会有任何迷茫感。新建一个连接的配置项如下Name连接名称随意填写比如“本地测试”。HostBroker地址格式是mqtt://127.0.0.1:1883。注意MQTTX要求带协议前缀本地没加密就用mqtt://TLS加密就填mqtts://。Username/Password如果Broker开了认证就填写。本地EMQX默认允许匿名可以留空。Client ID客户端唯一标识默认会自动生成建议保持唯一。如果两个客户端用同一个ClientID连接同一个Broker前者会被强制踢下线这个坑后面还会讲到。Clean Session是否清除会话测试时勾选即可。点击“连接”按钮中间面板显示“Connected”就说明连接成功了。然后订阅一个主题比如device/test再往这个主题发送一条JSON消息比如{temperature: 25.5}立刻就能在订阅端看到这条消息。整个流程不到一分钟非常直观。MQTTX还有一些高级功能也很实用。比如它可以在一个连接里同时订阅多个主题可以给服务端发送遗嘱消息测试离线感知可以复制一条消息重新编辑还有脚本功能能自定义消息生成规则模拟多设备上报。这些在调试阶段能省下大量时间。3. 若依后端集成MQTT完整实操3.1 引入依赖与自定义配置项我在若依后端项目的pom.xml里添加Paho客户端依赖。dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependency然后在application.yml中添加自定义的MQTT相关配置。因为若依框架本身有多环境配置application-dev.yml、application-prod.yml我把MQTT配置放在公共配置里方便不同环境覆盖。mqtt: host: tcp://127.0.0.1:1883 client-id: ruoyi-server-001 username: admin password: public topic: device/# qos: 1 completion-timeout: 3000 keep-alive-interval: 60参数含义说明hostBroker地址格式tcp://ip:port。client-id客户端唯一标识服务端实例多部署时要注意唯一否则会互相踢下线。username/password连接Broker的认证信息。topic服务端启动后默认订阅的主题可以用通配符。qos订阅的默认服务质量等级。keep-alive-interval心跳间隔单位秒客户端会在这个时间周期内发送PINGREQBroker连续一段时间没收到心跳就判定连接断开。3.2 客户端初始化连接参数、自动重连、主题订阅接下来创建配置类在Spring Boot启动时初始化MQTT客户端并建立连接。这一段是核心照着写就能跑通。package com.ruoyi.framework.mqtt; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; Data Component ConfigurationProperties(prefix mqtt) public class MqttProperties { private String host; private String clientId; private String username; private String password; private String topic; private int qos; private int completionTimeout; private int keepAliveInterval; }package com.ruoyi.framework.mqtt; import lombok.extern.slf4j.Slf4j; import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import javax.annotation.Resource; Slf4j Configuration public class MqttConfig { Resource private MqttProperties mqttProperties; Resource private MqttMessageCallback mqttMessageCallback; Bean public MqttClient mqttClient() throws Exception { MemoryPersistence persistence new MemoryPersistence(); MqttClient client new MqttClient( mqttProperties.getHost(), mqttProperties.getClientId(), persistence ); MqttConnectOptions options new MqttConnectOptions(); options.setUserName(mqttProperties.getUsername()); options.setPassword(mqttProperties.getPassword().toCharArray()); options.setCleanSession(true); options.setConnectionTimeout(mqttProperties.getCompletionTimeout()); options.setKeepAliveInterval(mqttProperties.getKeepAliveInterval()); // 开启自动重连 options.setAutomaticReconnect(true); // 设置遗嘱消息 options.setWill(device/server/status, offline.getBytes(), 1, false); client.setCallback(mqttMessageCallback); client.connect(options); if (client.isConnected()) { client.subscribe(mqttProperties.getTopic(), mqttProperties.getQos()); log.info(MQTT连接成功已订阅主题{}, mqttProperties.getTopic()); } return client; } }这里有三个点值得展开说。第一为什么用MemoryPersistencePaho客户端支持将消息持久化到磁盘保证客户端重启后未发送完的消息还能继续处理。但若依通常部署在标准化服务器环境里磁盘路径配置麻烦内存持久化简单可靠配合QoS 1已经能满足绝大多数业务。如果真有严格的消息补偿需求建议在业务层做幂等处理而不是依赖客户端持久化。第二为什么在订阅时用通配符device/#我按业务规划把所有设备相关消息都收敛到device这个父级主题下服务端一次订阅就能覆盖所有设备状态。通配符是很方便但要注意安全性和消息量如果所有消息都混在一个大主题下后端处理的压力会很大。所以规划主题时要有层级设计比如device/{deviceId}/status、device/{deviceId}/cmd/reply代码里再按topic段解析出设备ID。第三遗嘱消息的用途。我设了一条device/server/status的遗嘱消息内容是offline。如果若依服务端进程崩溃Broker会立刻把这个遗嘱消息发出去订阅这个主题的运维系统就能第一时间感知到服务端离线了。这是很实用的高可用设计平时不会有感知故障时能省下大量排查时间。3.3 消息回调处理解析数据、入库、推送消息回调是整个集成的核心枢纽。设备上报的所有数据都会汇聚到这里所以处理的逻辑要设计得清晰、可扩展。我的做法是回调里只做消息解析和分发具体的业务处理交给不同的Service。package com.ruoyi.framework.mqtt; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.ruoyi.system.service.IDeviceDataService; import com.ruoyi.system.service.IWebSocketService; import lombok.extern.slf4j.Slf4j; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttCallbackExtended; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.springframework.stereotype.Component; import javax.annotation.Resource; Slf4j Component public class MqttMessageCallback implements MqttCallbackExtended { Resource private IDeviceDataService deviceDataService; Resource private IWebSocketService webSocketService; Override public void connectComplete(boolean reconnect, String serverURI) { log.info(MQTT连接完成重连标志{}, reconnect); // 断线重连成功后需要重新订阅主题 // 实际项目中可以在这里重新订阅 } Override public void connectionLost(Throwable cause) { log.error(MQTT连接丢失, cause); // AutomaticReconnect开启后这里只记录日志重连由框架处理 } Override public void messageArrived(String topic, MqttMessage message) { String payload new String(message.getPayload()); log.info(收到消息主题{}内容{}, topic, payload); try { JSONObject json JSON.parseObject(payload); // 从主题中解析设备ID例如 device/CH001/status String[] topicParts topic.split(/); String deviceId topicParts.length 2 ? topicParts[1] : null; // 业务分发 if (topic.endsWith(/status)) { deviceDataService.saveDeviceStatus(deviceId, json); } else if (topic.endsWith(/alarm)) { deviceDataService.saveAlarm(deviceId, json); } // 推送给前端 webSocketService.sendMessage(topic, payload); } catch (Exception e) { log.error(MQTT消息处理失败, e); } } Override public void deliveryComplete(IMqttDeliveryToken token) { // 消息发布完成回调一般用于指令下发确认 } }消息处理里最重要的原则是回调方法里不要做耗时操作。因为MQTT的messageArrived是单线程回调如果这里去查数据库、调用外部接口、写日志文件消息处理的吞吐量会严重下降。我实际调试时发现当设备量大之后处理不过来会导致消息积压、延迟增大。正确做法是收到消息后立即放入线程池异步处理Resource private ThreadPoolTaskExecutor mqttTaskExecutor; Override public void messageArrived(String topic, MqttMessage message) { mqttTaskExecutor.execute(() - processMessage(topic, message.getPayload())); }这样消息到达后立即返回业务处理放到独立线程池中执行互不阻塞。线程池的配置可以在若依框架里直接复用现有的异步任务线程池也可以单独创建一个核心线程数和队列大小根据消息量来定我这次按每秒20条消息的业务量配了8个核心线程、队列容量200。3.4 动态主题订阅在若依后台维护设备主题有一种常见需求设备上线后后台管理系统需要给这台设备下发指令就涉及向指定主题发布消息。还有一种更细的场景不同设备类型只需要订阅对应的主题而不是所有设备。这时动态订阅机制就显得很重要。在若依框架里我做了这样一个设计设备表新增一个topic字段设备接入时后台新增设备并记录该设备的主题。启动时先订阅默认主题device/#接收所有消息同时通过若依的定时任务或者设备上线的接口调用订阅管理服务动态增加或减少订阅。package com.ruoyi.framework.mqtt; import lombok.extern.slf4j.Slf4j; import org.eclipse.paho.client.mqttv3.MqttClient; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; Slf4j Service public class MqttSubscribeService { Resource private MqttClient mqttClient; /** * 记录已订阅的主题避免重复订阅 */ private final SetString subscribedTopics ConcurrentHashMap.newKeySet(); public void subscribe(String topic, int qos) { if (subscribedTopics.contains(topic)) { log.info(主题 {} 已订阅过跳过, topic); return; } try { mqttClient.subscribe(topic, qos); subscribedTopics.add(topic); log.info(动态订阅主题成功{}, topic); } catch (Exception e) { log.error(动态订阅主题失败{}, topic, e); } } public void unsubscribe(String topic) { try { mqttClient.unsubscribe(topic); subscribedTopics.remove(topic); log.info(取消订阅主题成功{}, topic); } catch (Exception e) { log.error(取消订阅主题失败{}, topic, e); } } }这里用ConcurrentHashMap的keySet作为线程安全的订阅主题集合是为了避免并发场景下重复订阅的问题。在若依的Service层调用这个服务时你需要在业务Service里注入MqttSubscribeService在设备新增、删除的接口里加上对应的订阅和取消逻辑。但有一个容易忽略的点断线重连后Paho客户端不会自动恢复之前动态订阅的主题。自动重连机制只恢复连接不恢复会话和订阅。所以我在MqttMessageCallback的connectComplete方法里把MqttSubscribeService里维护的主题集合重新订阅一遍Override public void connectComplete(boolean reconnect, String serverURI) { log.info(MQTT连接完成重连标志{}重新订阅动态主题, reconnect); subscribedTopics.forEach(topic - { try { if (!mqttClient.isConnected()) { return; } // 实际项目中这里需要重新遍历订阅 // mqttClient.subscribe(topic, qos); } catch (Exception e) { log.error(重新订阅失败{}, topic, e); } }); }如果你在前面的阶段没有考虑这个细节生产环境一旦网络抖动触发重连部分设备消息就会静默丢失排查起来非常头疼。4. 前端实时展示Vue3中接入WebSocket4.1 后端提供WebSocket服务若依框架本身没有内置WebSocket支持需要自己加一个WebSocket服务端。好在Spring Boot对WebSocket支持很完善新建一个配置类启用WebSocket再写一个端点类处理连接和消息推送。package com.ruoyi.framework.websocket; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.config.annotation.WebSocketConfigurer; import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; import javax.annotation.Resource; Configuration EnableWebSocket public class WebSocketConfig implements WebSocketConfigurer { Resource private WebSocketServer webSocketServer; Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { registry.addHandler(webSocketServer, /ws/mqtt) .setAllowedOrigins(*); } }WebSocket服务端核心类package com.ruoyi.framework.websocket; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; import java.util.concurrent.ConcurrentHashMap; Slf4j Component public class WebSocketServer extends TextWebSocketHandler { private static final ConcurrentHashMapString, WebSocketSession SESSION_MAP new ConcurrentHashMap(); Override public void afterConnectionEstablished(WebSocketSession session) { SESSION_MAP.put(session.getId(), session); log.info(WebSocket连接建立{}当前连接数{}, session.getId(), SESSION_MAP.size()); } Override protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { // 接收客户端消息通常用于心跳或指令 } Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { SESSION_MAP.remove(session.getId()); log.info(WebSocket连接关闭{}, session.getId()); } public void sendToAll(String topic, String payload) { TextMessage message new TextMessage(payload); SESSION_MAP.forEach((id, session) - { try { if (session.isOpen()) { session.sendMessage(message); } } catch (Exception e) { log.error(WebSocket推送失败{}, id, e); } }); } }然后在之前的MQTT消息回调里把解析好的消息直接调用webSocketServer.sendToAll(topic, payload)推到前端。注意这里做了个简化处理没有区分用户权限。实际项目中建议结合若依的登录用户体系按用户订阅的设备主题做定向推送避免用户A看到用户B的设备数据。4.2 Vue组件中建立连接与消息渲染前端Vue这边比较简单组件挂载时创建WebSocket连接收到消息更新页面数据。export default { name: DeviceMonitor, data() { return { deviceStatusList: [], ws: null } }, created() { this.initWebSocket() }, beforeUnmount() { if (this.ws) { this.ws.close() } }, methods: { initWebSocket() { const protocol location.protocol https: ? wss : ws const wsUrl ${protocol}://${location.host}/ws/mqtt this.ws new WebSocket(wsUrl) this.ws.onopen () { console.log(WebSocket连接成功) } this.ws.onmessage (event) { const data JSON.parse(event.data) this.handleMqttMessage(data) } this.ws.onclose () { console.log(WebSocket连接断开3秒后重连) setTimeout(() { this.initWebSocket() }, 3000) } this.ws.onerror (error) { console.error(WebSocket连接错误, error) } }, handleMqttMessage(data) { // 根据设备ID更新列表 const index this.deviceStatusList.findIndex(item item.deviceId data.deviceId) if (index -1) { this.$set(this.deviceStatusList, index, data) } else { this.deviceStatusList.push(data) } } } }画面上用若依自带的Table组件展示设备列表用Tag组件显示在线状态、Alert组件显示告警再配一个ECharts折线图展示电压电流变化就是一个很完整的设备监控页面。前端这块本身不复杂真正需要注意的还是WebSocket的心跳与断线重连浏览器不会帮你维持连接网络抖动就可能断开所以我在onclose里做了3秒自动重连。不过这里有个坑如果你在MQTT消息回调里给前端推的是原始消息前端每次都要解析主题字符串、判断消息类型逻辑会越写越乱。建议后端在推送前先包装一层比如统一格式化成{topic: device/CH001/status, deviceId: CH001, type: status, data: {...}, timestamp: 1690000000000}前端拿到一个完整自洽的消息对象直接渲染就行。5. 常见问题与排查实录5.1 连接不上Broker怎么排查这个坑几乎每个新手都会踩一遍我把排查路径整理成清单出了问题按顺序对照现象可能原因解决办法connect timed out网络不通或Broker未启动先ping服务器IP再用telnet ip 1883测试端口Connection refusedBroker端口没监听或防火墙拦截检查EMQX是否启动确认1883端口监听Not authorized to connect用户名或密码错误在EMQX Dashboard里新建用户确认配置正确ClientId already in use有另一个客户端用了相同ClientID更换ClientID确保全局唯一MqttException: Unexpected errorbroker版本和客户端不兼容确认broker支持MQTT 3.1.1用mqttv3的兼容性最好排查连接问题有一个技巧先不管若依代码直接用MQTTX试着连接同一个Broker。MQTTX能连上说明Broker和网络没问题问题就出在若依后端的配置上MQTTX也连不上那就先解决Broker的问题。这个二分法能快速缩小范围。5.2 消息收不到/重复收到是怎么回事这类问题表象相似但原因差别很大要结合具体场景来判断。收不到消息先看三处第一Broker管理界面里能看到这个Topic的消息吗看不到说明设备根本没发上来。第二若依后端的订阅主题和设备发布的主题是否匹配device/001/status和device/001/state就差一个字符消息就是收不到。第三QoS等级是否设置合理有些设备端用QoS 0发送Broker到服务端的订阅也用QoS 0任何网络抖动都可能丢消息。对于关键业务尽量在两端都配置QoS 1。重复收到消息最常见的原因是QoS 1的“至少一次”投递语义本身就允许重复。解决方式不是去改QoS而是业务处理时做幂等。我在设备数据上报里加了消息ID字段每条消息携带唯一的messageId数据库里加唯一索引。消息重复到达时插入操作直接报唯一键冲突在代码里捕获这个异常并忽略即可。消息丢失除了网络问题还有一种隐蔽的原因客户端设置了Clean Session为true连接断开期间的离线消息全部丢弃。如果业务对消息完整性要求高需要设置Clean Session为false并开启持久化会话代价是Broker需要为每个客户端缓存未确认的消息内存开销要大一些。5.3 断线重连失效与ClientID冲突Paho客户端的AutomaticReconnect开启后断线后会自动尝试重连但在某些情况下还是不生效。比如网络长时间中断后重新恢复自动重连机制会一直尝试但没有退避策略频繁的重连请求会增加Broker压力。实际项目中建议自己实现重连逻辑在connectionLost回调里用指数退避算法1秒、2秒、4秒...最大60秒重连避免服务端和Broker都被拖垮。另一个经典问题是多实例部署时的ClientID冲突。若依微服务版如果同一服务部署了多个实例所有实例都用同一个ClientID连接同一个Broker后面连接的会顶掉前面的。解决方式是在配置里给ClientID加上实例标识比如ruoyi-server-${spring.cloud.client.ip-address}-${server.port}保证每个实例唯一。如果用若依微服务版还需要注意一个问题多个微服务如果都集成了MQTT客户端同一主题会被多个服务同时消费造成重复处理。这时候要么拆分主题每个服务只订阅自己关心的主题要么引入消息队列多个服务订阅同一个主题但消息只在其中一个服务中处理需要自行设计分布式锁或消费组机制。5.4 若依相关集成注意事项与避坑总结最后整理几个和若依框架结合时的特殊问题这些都是实际项目中容易忽略的。第一若依的权限控制要延伸到MQTT主题。若依框架对HTTP接口有完善的角色权限控制但MQTT是独立通道如果不做任何处理任何拿到Broker账号的人都能订阅设备主题、获取全部设备数据。这里建议做两层控制一是Broker层面的ACL按设备维度限制用户只能订阅自己的主题二是服务端在消息回调里增加校验检查设备ID是否与当前系统中的设备匹配。第二若依定时任务可以和MQTT结合做指令下发。比如运营人员每天凌晨通过后台设置充电策略到点后需要给设备下发指令。常规做法是写一个若依的定时任务到点后调用MqttPublishService向设备主题发布控制消息。Paho的MqttClient里发布消息很简单public void publish(String topic, String payload, int qos, boolean retained) { MqttMessage message new MqttMessage(payload.getBytes()); message.setQos(qos); message.setRetained(retained); mqttClient.publish(topic, message); }第三若依Vue3版本TS报错问题。新版若依Vue3TypeScript项目结构会多一些类型定义集成WebSocket时如果把事件数据直接当any类型处理编译时可能报TS错误。建议定义好接口类型比如MqttMessagePayload使用JSON.parse(JSON.stringify(data))做类型转换时注意类型断言。我在实际开发中就遇到过一个诡异的时间问题类型断言写得不严谨导致字符串当成对象处理页面直接白屏后来统一走接口类型定义加运行时校验才稳定下来。第四不要忘了给若依的系统监控模块加MQTT连接状态的监控。若依自带定时任务和系统监控页面可以在系统监控里增加一个MQTT连接状态指标通过Actuator暴露健康信息再配置告警。这样MQTT连接断开时运维能第一时间收到通知而不是等用户反馈“页面数据不动了”才发现问题。第五数据量上来后要做落库优化。设备状态消息往往频率很高如果每条消息都直接插入数据库数据库压力会非常大。我在这个项目里是先把原始数据存到Redis列表定时任务每隔10秒批量写入MySQL或者用ClickHouse这种时序数据库存海量设备数据。MySQL写不了高频数据这在物联网场景是个常识但很多第一次做的人都会踩这个坑。最后分享一个调试心得集成过程中最花时间的往往不是写代码而是定位“消息到底走到哪一步了”。这是所有消息系统调试的难点MQTT尤其如此——消息在经过设备、Broker、后端、WebSocket、前端五个节点后任何一个环节出错都会表现为“页面没数据”或“数据不对”。我的习惯是先在MQTTX里订阅后端要订阅的主题确认设备此时是否在发消息再确认后端日志里是否打印了messageArrived的收到记录接着确认WebSocket服务端SESSION_MAP里是否有前端连接最后看前端控制台WebSocket有没有收到消息。这一步一步往下排问题一定出现在最后一个正常节点的下游。只要保持这个排查逻辑再复杂的问题也能快速定位。另外再提一句调试MQTT时日志一定要打好。我强烈建议在消息处理的多个关键点打上不同级别的日志收到消息打DEBUG级别并带主题和内容摘要入库成功打DEBUG推送WebSocket打DEBUG异常打ERROR并带上完整异常栈。这样排查问题时日志就是你的王牌debug效率能翻好几倍。希望这篇文章能帮你少踩几个坑照着做完能跑通一个完整的若依MQTT项目然后再根据自己的业务场景去扩展优化。