简介这份资源是面向 Java 后端开发者与物联网入门者的 Spring Boot 整合 MQTT 实战示例工程围绕基于 TCP 的发布/订阅通信场景解决项目中如何快速接入 MQTT、完成消息收发的问题适合已具备 Spring Boot 基础、希望向 IoT 通信方向延伸的中级开发者参考。压缩包共 89 个文件约 78KB以 53 个 xmlMaven 依赖与 IDE 配置、10 个 java 源码与 10 个编译后的 class、8 个 yml 配置文件为主另含 properties 配置、jar 包及 iml、lst 等工程描述文件完整保留了可导入运行的工程结构。资源已有 709 人学习下载说明其在同类整合示例中具备一定参考价值。工程内提供 MQTT 客户端工厂配置、发布服务与订阅服务等核心模块读者可据此理解连接参数配置、主题订阅回调与消息发布流程并在此基础上扩展 REST 接口或业务事件触发快速搭建自己的设备通信骨架。1. 从一条设备上报消息说起这个 demo 到底解决什么问题设备端通过 MQTT 协议往 broker 发一条温湿度数据服务端要能实时收到、解析、落库还要能反向给设备下发指令——这套链路在物联网项目里几乎是标配。但真到动手时很多人卡在第一步Spring Boot 项目里怎么把 MQTT 客户端接进来连接怎么管、断线怎么重连、消息怎么分发到不同业务处理器。这个 demo 就是干这件事的它把 spring boot 与 mqtt 的整合拆成最小可运行单元覆盖连接配置、订阅、发布、消息回调几个关键环节适合刚接触 tcp 通信与消息中间件的后端开发者拿来当起点也适合需要给现有系统补一条设备通道的熟手直接改。它不追求大而全没有复杂的设备管理后台也没有花哨的前端核心就是一条能跑通的消息通路。你把它跑起来能看到客户端连上 broker、订阅主题、收到消息、打印日志再改几行就能接自己的业务。下面按「先看清结构、再跑起来、最后动手改」的顺序拆。2. 项目结构与技术选型一个 MQTT 客户端是怎么被组织起来的2.1 依赖与版本为什么选 Eclipse Paho 而不是自己写 socketSpring Boot 本身不内置 MQTT 支持整合方式通常是引入第三方客户端库再包一层配置。这个 demo 走的是最常见的一条路用 Eclipse Paho 的 Java 客户端作为底层通信实现Spring Boot 负责装配和生命周期管理。依赖大致是这几块!-- pom.xml 关键依赖 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter/artifactId /dependency !-- MQTT 客户端Paho 是 Java 生态里用得最广的实现 -- dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependency !-- 配置元数据提示可选但建议加 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-configuration-processor/artifactId optionaltrue/optional /dependency选 Paho 的理由很实际它把 MQTT 3.1.1 的连接、心跳、QoS、遗嘱消息都封装好了自己写 socket 去拼协议报文属于重复造轮子而且重连和心跳这些细节极容易翻车。版本上 1.2.5 是长期稳定的选择和 Spring Boot 2.x、3.x 都能配合。注意 Paho 的 API 是阻塞式的回调在独立线程里跑这一点后面讲消息处理时会反复提到。2.2 目录结构配置、客户端、回调三块分开一个干净的整合 demo包结构一般长这样src/main/java/com/example/mqttdemo/ ├── MqttDemoApplication.java # 启动类 ├── config/ │ ├── MqttProperties.java # 连接参数绑定 │ └── MqttConfig.java # 客户端 Bean 装配 ├── client/ │ └── MqttClientManager.java # 连接、订阅、发布封装 ├── handler/ │ └── MqttMessageHandler.java # 消息回调与业务分发 └── controller/ └── PublishController.java # 对外触发发布的接口 src/main/resources/ └── application.yml # broker 地址、主题、客户端 ID这个划分的逻辑是config只负责把 yml 里的参数变成对象并创建客户端client负责连接状态和收发动作handler只关心收到消息后干什么。三者职责不混后面要换 broker、加主题、改业务改动面都很小。很多新手把所有逻辑塞进一个类结果连接管理和业务处理缠在一起出问题根本定位不到是哪一层。2.3 连接参数application.yml 里每一项的含义配置项不多但每一项都影响连接行为写错一个就连不上或者频繁掉线mqtt: broker: tcp://127.0.0.1:1883 # broker 地址tcp 前缀表示普通 TCP 连接 client-id: demo-client-001 # 客户端唯一标识重复会导致互相踢下线 username: admin # 认证用户名broker 未开启认证可留空 password: public # 认证密码 default-topic: device//data # 默认订阅主题 是单层通配符 qos: 1 # 服务质量等级 0/1/2 keep-alive: 60 # 心跳间隔单位秒 clean-session: true # 是否清除会话影响离线消息保留 connection-timeout: 10 # 连接超时单位秒 automatic-reconnect: true # 断线自动重连开关client-id是最容易踩的坑同一个 broker 上两个客户端用相同 ID后连的会把先连的顶掉日志里表现为反复重连。clean-session设成 true 时断线期间 broker 不保留订阅关系和离线消息重连后要重新订阅设成 false 才能收到离线期间堆积的 QoS 1/2 消息但要求 client-id 固定。default-topic里的匹配一层#匹配多层写device//data能收到device/A/data和device/B/data但收不到device/A/temp/data。3. 核心代码拆解连接、订阅、发布、回调四段链路3.1 客户端 Bean 装配把连接参数变成可管理的单例配置类的作用是把 yml 参数绑定成对象再据此创建MqttClient并交给 Spring 管理Configuration EnableConfigurationProperties(MqttProperties.class) public class MqttConfig { Bean public MqttClient mqttClient(MqttProperties props, MqttMessageHandler handler) throws MqttException { // 内存持久化demo 够用生产可换 MqttDefaultFilePersistence 落盘 MqttClient client new MqttClient( props.getBroker(), props.getClientId(), new MemoryPersistence()); MqttConnectOptions options new MqttConnectOptions(); options.setUserName(props.getUsername()); options.setPassword(props.getPassword().toCharArray()); options.setCleanSession(props.isCleanSession()); options.setKeepAliveInterval(props.getKeepAlive()); options.setConnectionTimeout(props.getConnectionTimeout()); options.setAutomaticReconnect(props.isAutomaticReconnect()); // 注册回调收到消息后交给 handler 处理 client.setCallback(handler); client.connect(options); return client; } }这段代码的关键点有三个。第一MqttClient是线程安全的做成单例 Bean 没问题但不要在每次发消息时 new 一个那样连接数会爆炸。第二setCallback必须在connect之前调用否则连接建立后到达的消息可能漏掉回调。第三MemoryPersistence只把未确认消息放内存进程重启就丢demo 阶段无所谓真上生产要换成文件持久化否则 QoS 1 的「至少一次」语义在重启后就断了。MqttConnectOptions里setAutomaticReconnect(true)是省心开关Paho 会在断线后按退避策略重连但要注意自动重连成功后之前的订阅关系在cleanSessiontrue下已经没了需要重新订阅这一点后面避坑章节会展开。3.2 订阅与发布主题通配符和 QoS 怎么选连接建立后订阅动作通常放在启动完成时执行避免连接还没就绪就订阅Component public class MqttClientManager implements ApplicationRunner { private final MqttClient client; private final MqttProperties props; public MqttClientManager(MqttClient client, MqttProperties props) { this.client client; this.props props; } Override public void run(ApplicationArguments args) throws MqttException { // 启动后订阅默认主题QoS 与配置一致 client.subscribe(props.getDefaultTopic(), props.getQos()); } public void publish(String topic, String payload, int qos) throws MqttException { MqttMessage message new MqttMessage(payload.getBytes(StandardCharsets.UTF_8)); message.setQos(qos); // retainedfalse消息不保留给后续订阅者 message.setRetained(false); client.publish(topic, message); } }ApplicationRunner保证在 Spring 容器启动完成后才执行订阅比在构造函数里订阅稳妥。publish方法把 payload 转成字节数组注意统一用 UTF-8设备端如果发的是 GBK 就会乱码。QoS 的选择有讲究QoS 0 最多一次丢了就丢了适合高频传感器数据QoS 1 至少一次可能重复适合指令下发但要自己做幂等QoS 2 恰好一次开销最大一般只在计费等场景用。demo 默认 1 是个折中实际项目里按消息重要性分级设置。setRetained(false)表示这条消息不保留新订阅者收不到历史值。如果做设备状态同步把 retained 设成 true 能让新上线的服务立刻拿到最后一次状态这是个很实用的技巧。3.3 消息回调一个 handler 怎么把消息分发给不同业务回调是整合的核心Paho 在独立线程里调用messageArrived这里做的事越轻越好Component public class MqttMessageHandler implements MqttCallbackExtended { private static final Logger log LoggerFactory.getLogger(MqttMessageHandler.class); Override public void connectComplete(boolean reconnect, String serverURI) { // 自动重连成功后必须重新订阅否则收不到消息 log.info(MQTT connected, reconnect{}, uri{}, reconnect, serverURI); } Override public void messageArrived(String topic, MqttMessage message) { String payload new String(message.getPayload(), StandardCharsets.UTF_8); log.info(recv topic{}, qos{}, payload{}, topic, message.getQos(), payload); // 按主题层级路由到不同业务避免 if-else 堆成山 String[] parts topic.split(/); if (parts.length 3 device.equals(parts[0])) { String deviceId parts[1]; dispatch(deviceId, payload); } } private void dispatch(String deviceId, String payload) { // 实际项目这里丢线程池或消息队列回调线程不能阻塞 log.info(dispatch device{}, data{}, deviceId, payload); } Override public void connectionLost(Throwable cause) { log.warn(MQTT connection lost, cause); } Override public void deliveryComplete(IMqttDeliveryToken token) { // 发布确认QoS 1/2 时回调 } }实现MqttCallbackExtended而不是MqttCallback是为了拿到connectComplete回调——这是自动重连后重新订阅的挂载点很多人用基础接口就漏了这一步表现为「断线重连后收不到消息」。messageArrived里只做解析和路由真正的业务处理要丢到线程池因为 Paho 的回调线程是串行的一个消息处理慢了会堵住后续所有消息这是血泪经验。主题解析用split(/)按层级取设备 ID比正则轻量也符合 MQTT 主题的层级语义。3.4 对外发布接口用 HTTP 触发一条 MQTT 消息为了验证发布链路demo 通常带一个 REST 接口RestController RequestMapping(/mqtt) public class PublishController { private final MqttClientManager manager; public PublishController(MqttClientManager manager) { this.manager manager; } PostMapping(/publish) public String publish(RequestParam String topic, RequestParam String payload, RequestParam(defaultValue 1) int qos) throws MqttException { manager.publish(topic, payload, qos); return ok; } }调POST /mqtt/publish?topicdevice/A/cmdpayloadonqos1就能往指定主题发一条消息配合 MQTT 客户端工具订阅同一主题即可看到效果。这个接口在 demo 里是验证手段实际项目中发布动作往往由业务事件触发把manager.publish注入到对应服务里调用即可接口层可以保留做运维手动下发。4. 跑起来从零启动到看到第一条消息4.1 准备 broker本地起一个最省事的demo 本身不含 broker需要先有一个 MQTT 服务端。最省事的方式是用容器起一个# 启动一个本地 broker映射 1883 端口 docker run -d --name mqtt-broker \ -p 1883:1883 \ -p 9001:9001 \ eclipse-mosquitto:2起完后确认端口监听正常1883是 MQTT over TCP9001是 WebSocketdemo 只用前者。如果本机已有 broker 或公司有测试环境把application.yml里的broker改成对应地址即可注意前缀是tcp://不是http://这是新手常写错的地方。4.2 启动应用与验证链路改好配置后直接启动 Spring Boot观察日志里是否出现连接成功和订阅成功。验证分两步先测订阅用任意 MQTT 客户端工具往device/A/data发一条消息看应用日志有没有打印recv topic...再测发布调接口往device/A/cmd发消息用客户端工具订阅该主题看能否收到。# 用命令行客户端订阅验证应用发布是否成功 mosquitto_sub -h 127.0.0.1 -p 1883 -t device//cmd -v # 触发应用发布 curl -X POST http://localhost:8080/mqtt/publish?topicdevice/A/cmdpayloadonqos1订阅端能打印出device/A/cmd on就说明发布链路通了。反过来用mosquitto_pub往device/A/data发消息应用日志出现recv就说明订阅链路通了。两条都通这个 demo 的核心功能就算跑起来了。4.3 日志与断线重连的观察方法验证重连最直接的办法是把 broker 停掉再启动观察应用日志停掉后应出现connection lostbroker 恢复后应出现connectComplete reconnecttrue并且随后能继续收到消息。如果重连后收不到消息八成是没在connectComplete里重新订阅回到 3.3 节补上即可。这个观察方法比看代码猜要可靠得多建议每次改连接相关配置都走一遍。5. 避坑与排查这几处翻车点我踩过5.1 现象两个实例互相顶下线日志反复重连原因client-id写死成同一个值多个实例或多次启动用了相同 IDbroker 按 MQTT 规范只保留一个连接后连的顶掉先连的。解决client-id 加实例区分比如拼上主机名或随机后缀同时如果用了cleanSessionfalseclient-id 又必须稳定这时要保证同一逻辑客户端全局唯一不能多实例共用。5.2 现象断线重连成功但再也收不到消息原因cleanSessiontrue时 broker 不保留订阅关系自动重连只恢复连接不恢复订阅。解决在connectComplete回调里重新执行subscribe把订阅动作收敛到一个方法里首次连接和重连都调它避免两处逻辑不一致。5.3 现象消息处理慢后面消息堆积延迟越来越大原因messageArrived在 Paho 的单线程回调里执行里面做了数据库写入或远程调用阻塞了后续消息。解决回调里只做解析和投递业务处理丢线程池或消息队列同时给线程池设有界队列和拒绝策略防止内存被撑爆。5.4 现象中文 payload 收到是乱码原因发布端和订阅端字符集不一致常见是设备端用 GBK 编码而应用按 UTF-8 解析。解决协议层约定统一 UTF-8解析时显式指定StandardCharsets.UTF_8不要依赖平台默认编码new String(bytes)这种写法在跨平台时就是玄学。5.5 现象QoS 1 下同一指令被执行两次原因QoS 1 语义是「至少一次」网络抖动时 broker 会重发客户端可能收到重复消息。解决业务侧做幂等用消息里的业务唯一 ID 去重或者把关键指令的 QoS 提到 2。指望传输层保证不重复是不现实的幂等这层后悔药得自己留。6. 二次开发把它改成能接真实设备的骨架跑通之后这个 demo 离能用还有几步改造思路按优先级排。第一步是把连接参数外置到配置中心并支持多 brokerMqttProperties从单对象改成列表每个 broker 对应一个MqttClientBean用Bean的 name 区分注入。第二步是消息路由从if-else换成策略表用主题前缀映射到处理器接口public interface TopicHandler { String topicPattern(); // 如 device//data void handle(String topic, String payload); } // 启动时把所有实现收集成 Map按模式匹配分发 MapString, TopicHandler handlers handlerList.stream() .collect(Collectors.toMap(TopicHandler::topicPattern, h - h));这样加一类设备只需新增一个实现类不用动回调代码。第三步是补可观测性把连接状态、收发计数、重连次数打成指标出问题时不用靠翻日志猜。第四步是持久化把MemoryPersistence换成文件持久化并给未确认消息做落库补偿。验证改造是否成功我一般用三个动作停 broker 看重连和重订阅、发重复消息看幂等、压一批消息看回调线程是否被堵。这三个动作覆盖了连接、语义、性能三条最容易出事的线。有个习惯是从那以后我每次接 MQTT 都先把connectComplete里的重订阅写死再写业务——因为重连丢订阅这个坑太隐蔽日志上看连接是好的消息就是不来排查半天才发现是订阅没恢复。希望帮到你。本文还有配套的精品资源点击获取