企业官网建设流程全解析

去年在做一套设备管理后台时，我遇到了一个很典型的物联网接入需求：设备端统一用 MQTT 协议上报数据，而后端管理平台需要实时订阅这些消息，同时也要能向下发控制指令。管理平台是用若依框架搭的，Java 技术栈，于是“EMQX + Java 订阅发送 + 若依框架”就成了一条必须打通的链路。当时从 EMQX 部署、Java 客户端封装，到跟若依的权限、日志、异步处理做融合，前前后后踩了不少坑。这篇文章就把这套方案的完整落地过程写出来，重点不是贴一堆官方 Demo，而是把“连接怎么管理、订阅怎么设计、消息怎么和若依业务打通”这些真问题讲清楚。如果你正在用若依做物联网管理后台，或者在研究 MQTT 接入，这篇应该能帮你省下不少试错时间。

1. 选型与链路设计：EMQX 在若依物联网后台中的定位

1.1 先搞清楚消息是怎么流动的

MQTT 的模型很简单，就是一个发布订阅模式：EMQX 是消息代理（Broker），设备端和后端服务都作为客户端接入 Broker，通过主题（Topic）进行消息路由。设备往device/{deviceId}/report这个主题发布数据，后端服务订阅这个主题就能收到；后端要下发指令，就往device/{deviceId}/command发布消息，设备侧通过订阅这个主题接收。

在若依后台的语境下，这套链路通常对应这样的业务场景：

• 设备管理页面展示所有接入设备，需要实时显示在线状态；
• 设备上报的温度、电量、状态等数据，需要写入数据库并在前端展示；
• 运营人员通过后台页面点击“重启设备”“升级固件”，实际就是一个指令下发动作。

所以 EMQX 在这里扮演的不是“数据库”，而是一个高并发的消息通道。Java 后端既要做订阅者，也要做发布者。

1.2 为什么选 EMQX 而不是其他 Broker

市面上的 MQTT Broker 可选很多，Mosquitto、EMQX、HiveMQ、VerneMQ 各有侧重。我最终选 EMQX，主要看重这几点：

• 性能与连接数：EMQX 基于 Erlang/OTP 开发，单节点能撑百万级连接，对于绝大多数业务系统来说完全够用，而且集群扩展非常方便。
• 管理界面完善：Dashboard 里可以直接看连接数、订阅关系、消息流量，还能在线发消息调试，排查问题特别直观。
• 规则引擎和钩子：可以做消息重发布、数据持久化到数据库等操作，虽然我的方案里主要靠 Java 端处理，但多一个能力总不是坏事。
• 部署简单：Docker 一行命令就能起一个节点，部署成本极低。

如果你只是本地测试或者设备量很小，用 Mosquitto 也完全没问题，API 层面都是标准 MQTT 协议，Java 端代码不用改动。

1.3 EMQX 在若依项目里应该放在哪一层

这是很多人容易搞混的地方。EMQX 是独立部署的中间件，不是嵌在若依工程里的一个模块。若依工程和 EMQX 之间通过 MQTT 客户端通信。所以在整体架构上，若依是一个“超级客户端”，它既订阅感兴趣的 Topic，也向指定 Topic 发布消息。

具体到若依工程的模块划分，我当时是把 MQTT 相关的封装放在了ruoyi-common模块里，因为多个业务模块（设备管理、告警、系统监控）都可能用到订阅和发送能力。如果只在某个业务模块里用，放在对应模块的service包下也完全可以。这个选择看你的业务边界，没有绝对标准。

2. 环境搭建：Docker 部署 EMQX 与若依工程集成准备

2.1 Docker 方式快速部署 EMQX

我最早的测试环境是在一台 Linux 服务器上用 Docker 跑的，指令非常简单。生产环境建议用 5.x 或 4.x 的稳定版本，我当时用的是 4.4。

docker run -d --name emqx \ -p 1883:1883 \ -p 8083:8083 \ -p 8084:8084 \ -p 8883:8883 \ -p 18083:18083 \ emqx/emqx:4.4.14

端口说明：

• 1883：MQTT 协议端口，Java 端走这个；
• 8083：MQTT over WebSocket 端口，前端页面直连会用到；
• 8883：MQTT over SSL/TLS 端口；
• 18083：Dashboard 管理界面端口，浏览器打开http://服务器IP:18083，默认账号admin、密码public。

注意：生产环境部署后第一件事是改 Dashboard 密码，以及创建独立的 MQTT 用户名密码，不要用默认配置直接跑业务。

如果你手里有 N1 盒子这种 ARM 设备，也能跑 EMQX，镜像同样兼容 ARM 架构，只是性能上限比 x86 服务器低一些，适合个人项目或低并发场景。

2.2 在 Dashboard 里创建测试资源和账号

部署完成后，我习惯先在 Dashboard 里做两件事。

第一，创建专门用于后端服务的认证账号。进入“访问控制 → 用户管理”，添加一个用户名backend、密码自定义。EMQX 默认开启了匿名认证，出于安全考虑，生产环境建议关闭匿名访问，只允许通过账号认证的客户端连接。

第二，会先打开“WebSocket”页面，选一个测试客户端在线发几条消息，确认 Broker 本身工作正常。这一步能帮你把“EMQX 的问题”和“Java 端的问题”先隔离开，后面联调时少猜一半的坑。

2.3 若依工程引入 MQTT 客户端依赖

若依项目基于 Spring Boot，引入依赖很简单。我选择的是Eclipse Paho的 Java 客户端，这是最主流的选择，API 稳定，文档齐全。另外一个值得考虑的备选是 HiveMQ MQTT Client，API 更现代，支持 MQTT 5.0，两者选一个就行。

在ruoyi-common模块的pom.xml里引入：

<dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</artifactId> <version>1.2.5</version> </dependency>

之所以放在ruoyi-common而不是ruoyi-admin，是因为我希望所有业务模块都能复用 MQTT 能力，而不是把这段代码困在 Web 层里。

安装依赖时如果你遇到“error adding module to project: null”这类问题，通常不是 Maven 依赖本身的问题，而是 IDEA 在导入若依这种多模块工程时状态异常。把工程根目录下的.idea文件夹删掉，重新用 Maven 面板导入，或者把 IDEA 升级到较新版本，基本都能解决。

2.4 配置文件里的 MQTT 连接参数

我在若依的application.yml里加了这样一组配置：

mqtt: broker: tcp://localhost:1883 client-id: ruoyi-backend-001 username: backend password: your-password default-topic: device/+/report qos: 1 connect-timeout: 10 keep-alive: 60

这里有几个细节值得说。

client-id在同一个 Broker 上必须唯一，否则后连接的客户端会把先连接的踢下线。如果你部署了多个若依后端实例，一定要给每个实例配置不同的 clientId，比如用${spring.application.name}-${random.uuid}这种方式动态生成。

default-topic我用了device/+/report，其中+是 MQTT 通配符，代表任意一层。这样一个订阅就能覆盖所有设备的报数消息，不需要为每台设备单独订阅。

3. Java 客户端封装：基于 Paho 的 MQTT 连接与重连

3.1 为什么单独封装一层

直接用MqttClient也能工作，但若依项目里多个 Service 都要用，如果每个地方都写一遍连接逻辑，连接实例就会满天飞，后期维护就是灾难。单独封装一个连接管理组件，统一负责连接、重连、回调分发，业务层只关心“拿主题和数据”。

我封装了三个类：

• MqttProperties：读取配置；
• MqttConnectionManager：负责连接、订阅、取消订阅、断开；
• MqttMessageDispatcher：把收到的消息路由到不同的业务处理器。

3.2 MqttProperties 配置类

@Component @ConfigurationProperties(prefix = "mqtt") public class MqttProperties { private String broker; private String clientId; private String username; private String password; private String defaultTopic = "device/+/report"; private int qos = 1; private int connectTimeout = 10; private int keepAlive = 60; // getter / setter 省略 }

写清楚 getter/setter 就可以，若依项目里这种配置类的写法很常见。

3.3 MqttConnectionManager 连接管理

这是整个封装里最核心的部分。Paho 客户端本身支持自动重连，但很多业务场景要求断线之后重新订阅，这时候光靠setAutomaticReconnect(true)还不够，需要监听重连成功事件并恢复订阅。

@Component public class MqttConnectionManager implements MqttCallback { private static final Logger log = LoggerFactory.getLogger(MqttConnectionManager.class); private final MqttProperties properties; private final MqttMessageDispatcher dispatcher; private MqttClient client; private MqttConnectOptions connectOptions; private final Set<String> subscribedTopics = new ConcurrentHashMap<String, Boolean>().keySet(); public MqttConnectionManager(MqttProperties properties, MqttMessageDispatcher dispatcher) { this.properties = properties; this.dispatcher = dispatcher; init(); } private void init() { try { client = new MqttClient(properties.getBroker(), properties.getClientId()); connectOptions = new MqttConnectOptions(); connectOptions.setAutomaticReconnect(true); connectOptions.setCleanSession(true); connectOptions.setConnectionTimeout(properties.getConnectTimeout()); connectOptions.setKeepAliveInterval(properties.getKeepAlive()); connectOptions.setUserName(properties.getUsername()); connectOptions.setPassword(properties.getPassword().toCharArray()); client.setCallback(this); client.connect(connectOptions); log.info("MQTT连接成功, broker={}", properties.getBroker()); subscribe(properties.getDefaultTopic(), properties.getQos()); } catch (MqttException e) { log.error("MQTT客户端初始化失败", e); } } public void subscribe(String topic, int qos) { try { client.subscribe(topic, qos); subscribedTopics.add(topic); log.info("MQTT订阅成功, topic={}, qos={}", topic, qos); } catch (MqttException e) { log.error("MQTT订阅失败, topic={}", topic, e); } } public void unsubscribe(String topic) { try { client.unsubscribe(topic); subscribedTopics.remove(topic); } catch (MqttException e) { log.error("MQTT取消订阅失败, topic={}", topic, e); } } public void publish(String topic, String payload, int qos) { try { MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8)); message.setQos(qos); message.setRetained(false); client.publish(topic, message); } catch (MqttException e) { log.error("MQTT消息发送失败, topic={}, payload={}", topic, payload, e); } } @Override public void connectionLost(Throwable cause) { log.warn("MQTT连接断开, 等待自动重连, 原因: {}", cause == null ? "unknown" : cause.getMessage()); } @Override public void messageArrived(String topic, MqttMessage message) { String payload = new String(message.getPayload(), StandardCharsets.UTF_8); log.info("MQTT收到消息, topic={}, payload={}", topic, payload); dispatcher.dispatch(topic, payload); } @Override public void deliveryComplete(IMqttDeliveryToken token) { // 发布完成回调，QoS 1/2 下可以在这里做可靠性确认 } }

这段代码里有几个关键点，值得单独说。

第一，client.setCallback(this)必须在connect()之前调用。如果顺序反了，连接期间收到的消息会因为没有回调而丢失，而且断线重连后消息也可能无法正常派发。

第二，Paho 的自动重连只负责重建 TCP 连接和会话，不负责恢复订阅。尤其是cleanSession=true的前提下，重连后 Broker 不会记住你的订阅关系，必须重新调用subscribe()。我在connectionLost和重连成功之间没有预留钩子，为了稳妥，可以把subscribe的恢复逻辑放在一个定时检查任务里，或者把automaticReconnect关闭，自己处理重连逻辑，在重连成功后显式恢复全部订阅。

第三，subscribedTopics这个集合记录了当前所有订阅主题。断线重连后，遍历这个集合重新订阅：

public void reconnectAndResubscribe() { try { if (!client.isConnected()) { client.connect(connectOptions); } for (String topic : subscribedTopics) { client.subscribe(topic, properties.getQos()); } } catch (MqttException e) { log.error("重连订阅恢复失败", e); } }

这个方案虽然笨重，但可靠。如果你的应用里订阅关系是动态变化的，这个集合就是唯一的权威来源。

3.4 MqttMessageDispatcher 消息分发

消息到达后不能直接写业务代码，因为一个 MQTT 主题可能对应多种消息类型，而且后续还可能新增主题。用ApplicationEventPublisher把消息发布成 Spring 事件，就是一种很优雅的解耦方式。

@Component public class MqttMessageDispatcher { private static final Logger log = LoggerFactory.getLogger(MqttMessageDispatcher.class); private final ApplicationEventPublisher eventPublisher; public MqttMessageDispatcher(ApplicationEventPublisher eventPublisher) { this.eventPublisher = eventPublisher; } public void dispatch(String topic, String payload) { if (topic == null || payload == null) { return; } log.debug("分发MQTT消息, topic={}", topic); eventPublisher.publishEvent(new MqttMessageEvent(this, topic, payload)); } }

事件类：

public class MqttMessageEvent { private final String topic; private final String payload; public MqttMessageEvent(Object source, String topic, String payload) { this.topic = topic; this.payload = payload; } public String getTopic() { return topic; } public String getPayload() { return payload; } }

业务侧监听事件即可。比如在DeviceReportListener里写上@EventListener方法，实现对设备消息的处理：

@Component public class DeviceReportListener { private static final Logger log = LoggerFactory.getLogger(DeviceReportListener.class); @Autowired private DeviceDataService deviceDataService; @EventListener public void onMessage(MqttMessageEvent event) { String topic = event.getTopic(); String payload = event.getPayload(); // 解析 topic 获取设备ID // 解析 payload 得到业务数据 deviceDataService.saveReportData(deviceId, payload); } }

这个模式的优点是“消息来了，业务的各个模块各自认领”，MQTT 连接层完全不知道也不关心业务细节。

4. 订阅与发送业务落地：主题设计、动态订阅与指令下发

4.1 主题规划是第一步，也是最容易忽略的一步

很多项目在 MQTT 接入阶段没有认真设计主题结构，上来就是test、data、msg123这种一杆子打到底的主题。设备少的时候看不出问题，设备一多，主题和业务纠缠不清，维护成本极高。

我当时定的规范是：

层级之间用/分隔，每一层都代表一个明确的维度。订阅端尽量使用通配符，device/+/report订阅所有设备的上报数据，device/{deviceId}/report只订阅指定设备的数据。

4.2 在后端代码里处理动态订阅

有几种业务场景需要后端运行时动态新增订阅。比如用户在前端绑定了一台设备，后台需要实时接收该设备的告警信息。这时候不可能重启服务，必须调用subscribe()方法新增订阅。

在若依的 Service 里调MqttConnectionManager.subscribe(topic, qos)即可。但要注意一件事：动态订阅多了之后，要防重，防止同一主题被重复订阅多次。Paho 对重复订阅的处理是“覆盖 QoS”，不会报错，但会带来日志干扰和不必要的心智负担。封装的时候可以在subscribedTopics里做幂等判断。

public void subscribeIfAbsent(String topic, int qos) { if (subscribedTopics.contains(topic)) { return; } subscribe(topic, qos); }

4.3 指令下发流程与若依接口串联

指令下发这个场景特别容易踩坑，因为它是“反向”链路。设备上报是单向的、被动接收，指令下发需要从若依的后台接口发起，再经过 EMQX 转给设备。

我提供一下思路。先写一个下发接口，暴露给前端调用：

@RestController @RequestMapping("/iot/device") public class DeviceCommandController extends BaseController { @Autowired private MqttConnectionManager mqttConnectionManager; @Log(title = "设备指令下发", businessType = BusinessType.INSERT) @PostMapping("/command") public AjaxResult sendCommand(@RequestBody DeviceCommandRequest request) { String topic = "device/" + request.getDeviceId() + "/command"; String payload = JSON.toJSONString(request.getCommandData()); mqttConnectionManager.publish(topic, payload, 1); return success(); } }

这里有三个容易踩的坑。

第一，指令下发一定要记日志。用若依自带的@Log注解，操作记录自动落到sys_oper_log表，出问题时能回溯是谁在什么时间下发了什么指令。

第二，指令下发要做好幂等。如果设备离线，消息发出去没人消费，用户很可能再点一次按钮，结果设备恢复上线后又突然收到一堆重复指令。这个问题我们后面专门说。

第三，如果指令需要设备回复确认，建议设计一个requestId字段放在消息体里，设备回复时带上这个字段，后端才能做请求-响应对账。否则你只知道发出去了一条消息，不知道设备到底执行了没有。

4.4 和若依权限体系的衔接

若依的后台权限管理已经很完善，指令下发接口天然要走一遍它的认证和鉴权链路。这里额外提一点：MQTT 的 Topic 权限和若依的页面权限是两套体系，不要试图用若依的用户权限去管 Topic 的细粒度访问。Topic 的访问控制应该在 EMQX 的 ACL 配置里做，若依层面只需要保证“能访问这个菜单的人才能调用下发接口”。

简单做法是给每个业务角色分配一个或者一组 Topic 前缀，在若依的菜单配置里绑定操作按钮权限；EMQX 那边用同一个用户名密码连接，也就是同一个后端服务统一收发消息。如果业务上确实需要区分不同部门的设备，再考虑在 EMQX 里配置多套认证账号，并把账号和设备分组绑定。

5. 与若依框架的融合实战：Spring 容器、日志与异步处理

5.1 不要在 MQTT 回调线程里直接用 Autowired 的坑

这是新手必踩的一个大坑。Paho 的messageArrived回调是在 MQTT 客户端的线程里执行的，如果你的MqttConnectionManager是 Spring 管理的 Bean，那在回调里调用其他 Spring Bean 是没问题的，因为 Bean 本身已经在容器里了。真正的问题发生在另一个场景：很多人喜欢在MqttCallback里 new 一个业务对象，或者用静态方法，结果业务对象里的@Autowired全部是 null。

解决办法有两个：

• 方案一：让回调类本身成为 Spring Bean，通过构造器注入依赖，像我在MqttConnectionManager里注入MqttMessageDispatcher一样；
• 方案二：不要在回调里直接处理业务，通过ApplicationEventPublisher把消息抛给 Spring 事件机制，在监听器里使用@Autowired。

方案二是我推荐的做法，它把 MQTT 传输层和业务层彻底解耦了，后面替换客户端都不会影响业务代码。

5.2 异步处理：别让 MQTT 回调线程变成业务线程

如果每次收到消息都要写数据库或者调外部接口，直接在@EventListener里同步处理，就会拖慢 MQTT 回调线程。Paho 默认的线程池有限，一旦积压，新的消息就会延迟甚至超时。

我的做法是：在监听器方法上标注若依里已有的@Async注解，让消息处理进入独立的线程池。

@Async("threadPoolTaskExecutor") @EventListener public void onMessage(MqttMessageEvent event) { // 业务处理 }

若依框架里已经配置好了线程池，直接用即可。这里提示一句：如果同步处理失败要实时感知，就别用@Async；如果只是普通的日志记录和异步入库，用@Async非常合适。

5.3 操作日志：把 MQTT 动作纳入若依审计体系

除了指令下发通过@Log记录之外，我还建议把 MQTT 的异常事件（连接断开、重连失败、订阅失败）纳入系统的日志体系。因为这些事件不会直接体现在用户操作里，但对系统运维非常关键。可以在MqttConnectionManager的异常分支里加上if (log.isErrorEnabled())或者调用自定义的告警服务，把重要故障信息同步给运维人员。

5.4 消息处理与数据库写入的最终一致性

设备上报最频繁的场景就是大量数据直接入库。这里要注意不要每个消息都做一次数据库事务，否则 CPU 和数据库连接都会成为瓶颈。常见的做法是批量落库：监听器收到消息后放进一个内存队列，定时任务每 5 秒或者积累 100 条批量插入一次。

如果你用的是若依自带的SysJob定时任务，可以很方便地注册一个批量入库任务。不过要记得给队列设置上限，防止内存溢出。这是消息量大之后必须考虑的事情。

6. 踩坑记录与可靠性调优：真机测试中遇到的几个典型问题

6.1 断线重连后订阅丢失

这是我最开始遇到的第一个大问题。cleanSession设为 true 的情况下，客户端掉线后 Broker 会清除会话信息，重连成功后如果客户端不重新订阅，就什么消息都收不到。Paho 的setAutomaticReconnect(true)只保证连接恢复，不保证订阅恢复。

解决办法就是我前面写的subscribedTopics集合 + 重连恢复逻辑。如果你用的是cleanSession=false，Broker 会保存会话和订阅关系，但也会有副作用：离线期间 QoS 1/2 的消息会积压在 Broker 端，客户端重连后一次性全部推送过来，可能瞬间把服务打爆。物联网场景下，我建议还是用cleanSession=true，自己管理重连订阅。

6.2 QoS 选择和重复消息处理

消息的 QoS 是 MQTT 里最容易被误解的概念。

• QoS 0：最多一次，可能丢失；
• QoS 1：至少一次，可能重复；
• QoS 2：刚好一次，性能开销最大。

设备上报这种允许偶尔丢一次的数据，用 QoS 0 就够了。控制指令这种涉及实际操作的，建议用 QoS 1，并在业务层做幂等。为什么不用 QoS 2？因为 EMQX 和 Paho 的 QoS 2 协商流程挺复杂，实际场景下 QoS 1 + 业务幂等 比 QoS 2 更可靠，也更简单。

重复消息的幂等方案很简单：消息体里带上唯一的messageId，后端收到后先查 Redis 或者数据库是否存在这个 ID，存在就直接丢弃。

// 伪代码 String messageId = json.getString("messageId"); boolean exists = redisTemplate.hasKey("mqtt:msg:" + messageId); if (exists) { return; } redisTemplate.opsForValue().set("mqtt:msg:" + messageId, "1", 1, TimeUnit.DAYS);

6.3 服务重启时的消息丢失

若依后端重启期间，设备还在正常运行，持续往 EMQX 上报数据。如果用了cleanSession=true，这些消息 Broker 不缓存，直接丢弃，重启后中间这段数据就没了。

这个问题要根据业务容忍度来权衡。少量设备或者低频上报场景，丢失一段数据问题不大。如果数据不能丢，可以考虑两个方案：

• 方案一：cleanSession=false，让 Broker 在客户端离线期间缓存 QoS 1/2 消息，但要注意积压上限；
• 方案二：EMQX 规则引擎直接把消息同时写一份到数据库，Java 端只负责实时业务。

我当时采用的是第二个方案，让 EMQX 通过规则引擎把上报数据落一份到 InfluxDB，Java 端只负责实时告警和状态更新。这样数据链路更稳，实时性和完整性都兼顾了。

6.4 消息格式与编码一致性

MQTT 的消息体本质上是字节数组，不限制格式。但实践中如果不统一用 UTF-8 编码 JSON，很容易出现中文乱码和解析失败。

我的建议是：设备端上报统一用 UTF-8 编码的 JSON 字符串，Java 端接收时统一用new String(message.getPayload(), StandardCharsets.UTF_8)。发送指令时同理，payload.getBytes(StandardCharsets.UTF_8)。不要依赖默认编码，不同操作系统的默认编码可能不一样。更规范的做法是在配置文件里约定好spring.messages.encoding=UTF-8，并让设备端固件也按这个标准来。

6.5 多实例部署时的重复消费

如果你对若依后端做了负载均衡，部署了两个实例，两个实例都订阅了device/+/report，那么一条设备上报消息会被两个实例同时收到。

MQTT 本身的机制就是这样，相同 clientId 的多个连接会互相踢下线，不同 clientId 的多个连接各自独立订阅。所以同一个主题被多个客户端订阅，Broker 就会向每个订阅者都分发消息。

如果系统里有些业务必须保证“一条消息只被处理一次”，有几个解决思路：

• 思路一：共享订阅，EMQX 支持$share/{group}/device/+/report，同组的客户端负载均衡地消费消息；
• 思路二：分布式锁或者 Redis 去重，保证同一业务只执行一次；
• 思路三：消息入队后再通过业务层面去重。

对于多数管理后台来说，思路二最实用。用 Redis 对业务数据唯一键加锁，重复消息直接跳过。

回到最开始的问题。如果你只是跑通一个 Demo，把 EMQX 部署起来，Java 端连上去订阅发送消息，其实一天就能搞定。但如果要把这套链路真正嵌入若依这种业务系统，并让它稳定跑在生产环境，就需要考虑连接管理、订阅恢复、消息幂等、权限关联、日志审计这些东西。

我个人在实际项目里最有体会的一点是：MQTT 接入本身不难，难的是消息进来之后如何与业务系统优雅地融合。用 Spring 事件解耦、统一主题规范、合理选择 QoS、把异常纳入若依日志体系，这几件事做好了，后面的扩展就会很顺。希望这篇记录能给你省点时间和精力。