简介：这份资源是面向 Java 后端开发者与物联网入门者的 Spring Boot 整合 MQTT 实战示例工程，围绕基于 TCP 的发布/订阅通信场景，解决项目中如何快速接入 MQTT、完成消息收发的问题，适合已具备 Spring Boot 基础、希望向 IoT 通信方向延伸的中级开发者参考。压缩包共 89 个文件，约 78KB，以 53 个 xml（Maven 依赖与 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 而不是自己写 socket
Spring Boot 本身不内置 MQTT 支持，整合方式通常是引入第三方客户端库再包一层配置。这个 demo 走的是最常见的一条路：用 Eclipse Paho 的 Java 客户端作为底层通信实现，Spring Boot 负责装配和生命周期管理。依赖大致是这几块：
<!-- pom.xml 关键依赖 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <!-- MQTT 客户端，Paho 是 Java 生态里用得最广的实现 --> <dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</artifactId> <version>1.2.5</version> </dependency> <!-- 配置元数据提示，可选但建议加 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-configuration-processor</artifactId> <optional>true</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 会在断线后按退避策略重连，但要注意：自动重连成功后，之前的订阅关系在cleanSession=true下已经没了，需要重新订阅，这一点后面避坑章节会展开。
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); // retained=false，消息不保留给后续订阅者 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?topic=device/A/cmd&payload=on&qos=1就能往指定主题发一条消息，配合 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 TCP，9001是 WebSocket，demo 只用前者。如果本机已有 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?topic=device/A/cmd&payload=on&qos=1'订阅端能打印出device/A/cmd on就说明发布链路通了。反过来，用mosquitto_pub往device/A/data发消息，应用日志出现recv就说明订阅链路通了。两条都通，这个 demo 的核心功能就算跑起来了。
4.3 日志与断线重连的观察方法
验证重连最直接的办法是把 broker 停掉再启动，观察应用日志：停掉后应出现connection lost，broker 恢复后应出现connectComplete reconnect=true，并且随后能继续收到消息。如果重连后收不到消息，八成是没在connectComplete里重新订阅，回到 3.3 节补上即可。这个观察方法比看代码猜要可靠得多，建议每次改连接相关配置都走一遍。
5. 避坑与排查：这几处翻车点我踩过
5.1 现象：两个实例互相顶下线，日志反复重连
原因：client-id写死成同一个值，多个实例或多次启动用了相同 ID，broker 按 MQTT 规范只保留一个连接，后连的顶掉先连的。解决：client-id 加实例区分，比如拼上主机名或随机后缀，同时如果用了cleanSession=false，client-id 又必须稳定，这时要保证同一逻辑客户端全局唯一，不能多实例共用。
5.2 现象：断线重连成功，但再也收不到消息
原因：cleanSession=true时 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 离能用还有几步，改造思路按优先级排。第一步是把连接参数外置到配置中心并支持多 broker，MqttProperties从单对象改成列表，每个 broker 对应一个MqttClientBean，用@Bean的 name 区分注入。第二步是消息路由从if-else换成策略表，用主题前缀映射到处理器接口：
public interface TopicHandler { String topicPattern(); // 如 device/+/data void handle(String topic, String payload); } // 启动时把所有实现收集成 Map，按模式匹配分发 Map<String, TopicHandler> handlers = handlerList.stream() .collect(Collectors.toMap(TopicHandler::topicPattern, h -> h));这样加一类设备只需新增一个实现类，不用动回调代码。第三步是补可观测性，把连接状态、收发计数、重连次数打成指标，出问题时不用靠翻日志猜。第四步是持久化，把MemoryPersistence换成文件持久化，并给未确认消息做落库补偿。
验证改造是否成功，我一般用三个动作：停 broker 看重连和重订阅、发重复消息看幂等、压一批消息看回调线程是否被堵。这三个动作覆盖了连接、语义、性能三条最容易出事的线。
有个习惯是从那以后我每次接 MQTT 都先把connectComplete里的重订阅写死，再写业务——因为重连丢订阅这个坑太隐蔽，日志上看连接是好的，消息就是不来，排查半天才发现是订阅没恢复。希望帮到你。
本文还有配套的精品资源，点击获取
