企业官网建设流程全解析

简介：这份资源是面向 Java 后端开发者与物联网入门者的 Spring Boot 整合 MQTT 实战示例工程，围绕基于 TCP 的发布/订阅通信展开，帮助读者理解如何在 Spring Boot 项目中接入 MQTT 协议、完成消息收发的基本框架搭建，适合已具备 Spring Boot 基础、希望快速上手 IoT 通信场景的中级开发者参考。压缩包共 89 个文件，以 53 个 xml 构建与依赖描述文件、10 个 java 源码、10 个编译后的 class、8 个 yml 配置及 properties 配置为主，另含少量 jar 与工程元数据文件，整体约 78KB，目录结构完整、便于直接导入 IDE 运行调试。目前已有 709 人学习下载。工程内提供了 MQTT 客户端工厂配置、发布服务与订阅服务的示例代码，读者可据此对照理解连接参数配置、主题订阅回调与消息发布流程，并在此基础上扩展 REST 接口或业务事件触发，适配更复杂的物联网通信需求。

1. 一个 Spring Boot 项目接 MQTT，为什么我建议先跑通这份 demo

设备侧上报一条温度，服务端要立刻收到并落库，中间还隔着 Broker、Topic、QoS 和心跳。很多团队第一次做物联网接入，代码写了两天，卡在“连上了但收不到消息”或者“断线后不重连”上。这份 Spring Boot 整合 MQTT 的 demo，解决的就是从零把 Java 服务端接进 MQTT 消息链路这件事：用 Spring Boot 起一个可运行的服务，通过 MQTT 客户端连上 Broker，完成订阅、发布、回调处理的最小闭环。它适合正在做设备接入、消息推送、边缘数据采集的后端同学，也适合想搞清楚 MQTT 和普通 TCP 通信差别的开发者。下面我按“先跑通、再改参数、最后排坑”的顺序拆，代码可以直接抄。

2. 先把链路跑通：Spring Boot 接 MQTT 的最小可运行结构

2.1 依赖与连接参数怎么定

MQTT 本质是跑在 TCP 之上的应用层协议，所以你会看到配置里既有 Broker 地址端口，又有 clientId、keepAlive 这类参数。Spring Boot 官方没有内置 MQTT starter，社区常见做法是引入 Eclipse Paho 客户端，再自己包一层配置。依赖只需要两个：Spring Boot Web（方便起服务验证）和 Paho。

<!-- pom.xml 关键依赖 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</artifactId> <version>1.2.5</version> </dependency>

Paho 1.2.5 是长期稳定的版本，API 变动小，网上资料多，出问题好查。这里不引 Spring Integration MQTT，是因为那份 demo 走的是轻量路线，直接操作MqttClient更透明，参数改起来心里有数。连接参数建议单独放application.yml，别硬编码：

mqtt: broker: tcp://127.0.0.1:1883 client-id: springboot-demo-server username: demo password: demo123 default-topic: device/+/telemetry qos: 1 keep-alive: 60 clean-session: true

clientId在同一 Broker 上必须唯一，重复会导致互相踢下线，这是后面避坑章要重点说的。default-topic里的+是单层通配符，device/+/telemetry能匹配device/A01/telemetry，但匹配不到device/A01/room1/telemetry，多层要用#。QoS 先设 1，保证至少送达一次，调试阶段够用。

2.2 客户端初始化与订阅回调

连接动作放在一个@Configuration里，用@PostConstruct在启动时建立连接并订阅。核心是MqttClient的构造、MqttConnectOptions的参数装配，以及setCallback注册消息到达后的处理逻辑。

@Configuration public class MqttConfig { @Value("${mqtt.broker}") private String broker; @Value("${mqtt.client-id}") private String clientId; @Value("${mqtt.default-topic}") private String topic; @Value("${mqtt.qos}") private int qos; private MqttClient client; @PostConstruct public void init() throws MqttException { // memory 持久化仅适合 demo，生产要换文件或数据库持久化 client = new MqttClient(broker, clientId, new MemoryPersistence()); MqttConnectOptions options = new MqttConnectOptions(); options.setUserName("demo"); options.setPassword("demo123".toCharArray()); options.setCleanSession(true); // 断线后不保留会话 options.setKeepAliveInterval(60); // 60 秒心跳 options.setAutomaticReconnect(true); // 自动重连，关键 client.setCallback(new MqttCallbackExtended() { @Override public void connectComplete(boolean reconnect, String serverURI) { // 重连成功后必须重新订阅，cleanSession=true 时尤其重要 try { client.subscribe(topic, qos); } catch (MqttException e) { e.printStackTrace(); } } @Override public void messageArrived(String t, MqttMessage msg) { System.out.println("topic=" + t + " payload=" + new String(msg.getPayload())); } @Override public void connectionLost(Throwable cause) { System.out.println("连接断开: " + cause.getMessage()); } @Override public void deliveryComplete(IMqttDeliveryToken token) { } }); client.connect(options); client.subscribe(topic, qos); } }

这段代码有三个点值得说清。第一，MqttCallbackExtended比基础MqttCallback多了connectComplete，自动重连成功后能在这里补订阅，否则重连了也收不到消息，这是最常见的翻车点。第二，setAutomaticReconnect(true)让 Paho 自己处理断线重连，但重连间隔是内置的退避策略，想自定义得实现MqttReconnectCallback。第三，MemoryPersistence只适合 demo，进程一挂消息就丢，生产环境要换成MqttDefaultFilePersistence或配合 Broker 的持久会话。

2.3 发布端与消息体约定

订阅跑通后，发布端用来验证链路。写一个 REST 接口，收到请求就往指定 Topic 发一条消息，方便用 curl 或 Postman 触发。

@RestController @RequestMapping("/mqtt") public class MqttController { @Autowired private MqttClient client; @PostMapping("/publish") public String publish(@RequestParam String topic, @RequestParam String payload, @RequestParam(defaultValue = "1") int qos) throws MqttException { MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8)); message.setQos(qos); message.setRetained(false); // 不保留，避免新订阅者收到历史脏数据 client.publish(topic, message); return "ok"; } }

消息体建议统一成 JSON，字段至少包含设备号、时间戳、指标值，别直接发裸字符串。原因很简单：后面要落库、要做规则判断，裸字符串解析成本高。setRetained(false)是刻意设的，retained 消息会让新订阅者立刻收到最后一条历史消息，调试时容易误以为“消息一直在发”，实际是 Broker 缓存的。发布端和订阅端用同一个MqttClient实例在 demo 里没问题，但生产环境建议收发分离，避免一个连接上的阻塞影响另一个方向。

3. QoS、会话与心跳：三个参数决定消息到底丢不丢

3.1 QoS 0/1/2 的实际差别

很多人以为 QoS 是“可靠等级”，设高就万事大吉，实际它换的是开销和重复。QoS 0 最多一次，发出去不管，网络一抖就丢；QoS 1 至少一次，靠 PUBACK 确认，但可能重复；QoS 2 恰好一次，四次握手，开销最大。设备上报温度这种场景，丢一条问题不大，QoS 0 就够；但如果是计费或告警，QoS 1 更合适，重复在业务侧用消息 ID 去重。

改 QoS 的位置有两处：订阅时的subscribe(topic, qos)和发布时的message.setQos(qos)。注意订阅 QoS 和发布 QoS 取两者较小值生效，别只改一边就以为生效了。

3.2 cleanSession 与持久会话的取舍

cleanSession=true表示每次连接都是全新会话，Broker 不保留订阅关系和离线消息；false则保留，断线期间 QoS 1/2 的消息会在重连后补发。demo 里设 true 是为了调试干净，但生产环境如果要求“设备离线期间的消息不能丢”，必须设 false，同时 clientId 要固定，不能每次启动随机生成。

// 持久会话配置示例 options.setCleanSession(false); options.setAutomaticReconnect(true); // clientId 必须稳定，建议用业务标识拼接，而非 UUID

设了 false 之后，重连时不要在connectComplete里无脑重复订阅，虽然重复订阅不会报错，但会让 Broker 侧订阅计数混乱。稳妥做法是用一个标志位判断首次连接还是重连。

3.3 keepAlive 与 TCP 层的配合

keepAlive 是客户端告诉 Broker“我最多多久没消息就算掉线”的秒数，Paho 会在空闲时发 PINGREQ。设太短，网络稍抖就频繁重连；设太长，Broker 要等很久才判定掉线。60 秒是常见起点，移动网络或弱网环境可以放到 120。这里和普通 TCP 通信的差别在于：TCP 的 keepalive 是内核级、默认两小时，MQTT 的 keepAlive 是应用级、秒级可控，所以做设备接入时别指望 TCP 层帮你探活。

4. 避坑与排查：连不上、收不到、重复收的现场记录

4.1 现象：启动就报“Connection refused”，Broker 明明在跑

原因通常是地址协议头写错或端口不对。Paho 要求tcp://host:port，写成http://或漏了tcp://都会连不上；另外 1883 是明文端口，8883 是 TLS 端口，混用直接拒绝。解决：先用telnet 127.0.0.1 1883确认端口通，再核对配置里的协议头，TLS 场景还要额外配SSLSocketFactory。

4.2 现象：两个服务同时启动，其中一个反复掉线

原因就是 clientId 冲突。同一 Broker 上相同 clientId 的后连接会把前一个踢掉，日志里表现为connectionLost反复触发。解决：clientId 加业务前缀或实例后缀，比如server-${spring.application.name}-${random.uuid}，但注意用了持久会话就不能随机，得用稳定的业务标识。

4.3 现象：断网恢复后日志显示重连成功，但一条消息都收不到

原因是cleanSession=true时重连后订阅关系已丢，而代码只在启动时订阅了一次。解决：在connectComplete回调里重新subscribe，这是MqttCallbackExtended存在的意义。如果用的是基础回调，就得自己在重连逻辑里补订阅。

4.4 现象：同一条消息被处理了两次

QoS 1 的“至少一次”语义决定的，网络抖动导致 PUBACK 丢失，Broker 会重发。解决：业务侧做幂等，消息体里带唯一 messageId，处理前查重；或者对时效性要求不高的场景降到 QoS 0。

4.5 现象：本地能连，部署到服务器就超时

常见于 Broker 只监听了 127.0.0.1，或者服务器安全组没放行 1883。解决：确认 Broker 配置里listener绑定的是0.0.0.0，再检查防火墙和端口放行。这类问题排查顺序永远是“本机端口 → 网络可达 → 认证 → 订阅”，别一上来就怀疑代码。

5. 从 demo 到能用：消息落库、异常兜底与压测验证

demo 跑通只是起点，真正上线前我会补三件事。第一件是消息落库，在messageArrived里别直接写业务逻辑，先丢进一个有界队列，由独立线程消费入库，避免回调线程被慢 SQL 拖死。第二件是异常兜底，messageArrived里抛异常会导致 Paho 内部状态异常，务必 try-catch 包住，记录原始 payload 再决定丢弃还是重试。第三件是压测验证，用脚本按每秒 N 条往 Topic 灌消息，观察消费延迟和内存增长，QoS 1 下重点看重复率。

// 回调里只做入队，业务异步处理 private final BlockingQueue<String> queue = new LinkedBlockingQueue<>(10000); @Override public void messageArrived(String t, MqttMessage msg) { String body = new String(msg.getPayload(), StandardCharsets.UTF_8); if (!queue.offer(body)) { // 队列满说明消费跟不上，记录并丢弃，避免 OOM System.err.println("queue full, drop: " + body); } }

队列容量要按峰值流量估，满了宁可丢也不能阻塞回调线程。压测时我会盯三个指标：入队成功率、消费延迟 P99、重连次数。重连次数异常升高，八成是 keepAlive 太短或网络质量差，而不是代码问题。

还有一个容易被忽略的点：Topic 设计。device/+/telemetry这种通配订阅在设备量大时会让单个客户端扛不住，常见做法是按业务域拆 Topic，服务端用多个订阅分摊，或者引入共享订阅（Broker 支持的话）做负载均衡。这份 demo 用的是单客户端单订阅，适合验证链路，不适合直接扛几千台设备。

从那以后我每次接 MQTT 项目，都会先把connectComplete里的重订阅、clientId 唯一性、QoS 与幂等这三件事写成检查项过一遍，再谈业务逻辑。希望帮到你。

本文还有配套的精品资源，点击获取