直接上结论在 Spring Boot 项目里做 MQTT 通信最稳的方案是用spring-integration-mqtt它可以无缝融入 Spring 的事件机制和依赖注入体系比单独用 Paho 客户端再自己写线程管理要省心得多。这篇文章我尽量把从环境准备、代码实现、配置解读到线上踩坑的经验一次讲完适合那些刚准备在 Spring Boot 里接入 MQTT、对订阅和发布机制还不太熟的开发者参考。1. 为什么是 MQTT Spring Boot物联网接入场景下的选型思考做后端开发的朋友对 MQTT 这个名字应该都不陌生但真正动手做过的人会发现它和 HTTP 接口的思路完全不一样。HTTP 是请求-响应模型客户端主动拉数据而 MQTT 是发布-订阅模型设备端把数据推送到 broker服务端订阅对应的主题就能持续收到消息。这个特性决定了它在物联网场景下的天然优势——适合低带宽、弱网、设备数量大的环境。Spring Boot 在这套体系里扮演的角色其实是集成层。设备侧只认 MQTT 协议不关心你后端是 Java、Go 还是 Node.js。Spring Boot 要做的就是稳定地连接 broker、承载订阅回调、把消息转化成业务事件。选 Spring Boot 而不是裸写 Paho 客户端核心理由有三个依赖注入方便。连接配置、主题参数、消息处理器都能定义成 Bean测试和替换成本低。框架自带生命周期管理。应用启动时自动建立连接销毁时自动释放资源不用自己钩 Spring 的回调接口写一堆样板代码。消息驱动的模型天然衔接。把收到的 MQTT 消息转成 Spring 的ApplicationEvent业务代码只需要监听事件和写普通的异步任务没区别学习曲线平缓。我见过不少项目一开始用原生的 Eclipse Paho Java 客户端确实能跑但一旦涉及多个主题订阅、连接断开重连、消息幂等这些现实问题时代码会迅速膨胀。与其自己造轮子不如直接用 Spring Integration 的 MQTT 适配器。它底层的 MQTT 协议实现也是 Paho但对外提供的是 Spring 风格的消息通道处理逻辑清晰很多。2. 环境准备Broker 安装与 Spring Boot 版本选择的常见坑2.1 本地开发用哪个 Broker 最顺手开发调试阶段Broker 的选择直接决定了你排错效率。主流选项有 Mosquitto、EMQX、HiveMQ 这几种我个人的建议是本地开发优先用 Mosquitto生产环境再评估 EMQX 这类支持集群的 Broker。Mosquitto 的安装没什么门槛以 Windows 为例下载安装包后完成安装服务默认是自启动的。想测试 Broker 状态时最直接的方式就是手动订阅一个主题看有没有报错。下面这两个命令是开发的标配# 订阅 test/topic 主题的所有消息 mosquitto_sub -h localhost -p 1883 -t test/topic # 向 test/topic 发布一条消息 mosquitto_pub -h localhost -p 1883 -t test/topic -m hello from shell如果你不想用命令行建议装一个 MQTTX 桌面客户端。它能让你可视化地查看所有主题的消息流调试阶段比命令行高效太多。我经常是先开 MQTTX 订阅所有相关主题用#通配符然后调后端的发布逻辑消息有没有发出去、payload 对不对一眼就能确认。2.2 Spring Boot 版本和 MQTT 依赖的配套关系这个环节是最容易踩坑的地方尤其是刚上手的朋友看到依赖报错就会慌。先说结论Spring Boot 2.x 和 3.x 在 MQTT 依赖的引入方式上有差异一个是 javax一个是 jakarta这不是单纯换个坐标的问题。如果你用的是 Spring Boot 2.7.x引入下面这些依赖dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-integration/artifactId /dependency dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId /dependency如果你用的是 Spring Boot 3.x依赖本身不需要变Spring Integration 6.x 依然提供了spring-integration-mqtt但编译时如果报jakarta.*相关的包缺失你需要额外确认项目的 JDK 版本是 17并确保 Maven 私服能拉到 Spring Integration 6.0 以上的版本。本质上这不是 MQTT 的问题而是整个 Spring Boot 3.x 对 Jakarta EE 9 的迁移导致的连带影响。还有个更隐蔽的问题如果你只在 pom.xml 里加了spring-integration-mqtt而不加spring-boot-starter-integration可能会在运行时报MessageChannel相关的类找不到。因为 MQTT 适配器依赖了 Spring Integration 的核心模块这个核心模块不会自动传递进来必须显式声明。这个坑我踩过所以这里单独拎出来提醒一句。另外关于热搜词里提到的springboot版本太高我补充一种你没想过的场景Spring Boot 3.x 默认的 HTTP 客户端和 Web 容器都变了服务器上如果还跑着 JDK 8启动直接报UnsupportedClassVersionError。这不算 MQTT 的问题但项目升级前必须确认运行环境的 JDK 版本。MQTT 模块本身不挑 JDK 版本是 Spring Boot 全家桶在挑。3. 配置文件详解连接参数、主题定义与线程池设置配置文件写得好不好直接影响后续的维护成本。MQTT 的配置项虽然不多但每个字段的含义必须清楚。下面是我在项目里常用的一套配置模板你可以直接抄spring: mqtt: broker-url: tcp://localhost:1883 client-id: springboot-mqtt-client username: admin password: admin123 default-topic: test/topic completion-timeout: 3000 # 连接参数 keep-alive-interval: 60 automatic-reconnect: true clean-session: true max-inflight: 100注意Spring Boot 官方并没有提供spring.mqtt这个自动配置项——这是该项目的自定义配置前缀。因为 Spring Boot 的 MQTT 自动配置长期以来都是缺失的必须自己封装一个MqttProperties配置类来读取这些字段。这也解释了一个通用现象你搜索 Spring Boot MQTT 教程时每个教程里的配置类写法都不太一样本质都是各自封装。具体到每个参数的含义client-id客户端在 broker 上的唯一标识。同一时刻如果两个客户端使用同一个 client-id 连接同一个 broker前一个会被强制断开。这在多实例部署时必须通过配置差异化比如加上应用名或实例序号来避免互踢。keep-alive-interval心跳间隔。设备弱网环境下建议设置 30-60 秒太短会增加无效网络包太长则会让 broker 误判设备在线。automatic-reconnect这个参数非常关键建议永远设为true。如果连接因网络波动断开Paho 客户端会自动发起重连而业务代码完全不用感知。clean-session如果为truebroker 不会保留该客户端的会话状态断线期间漏掉的消息不会再补推如果为false能接收断线期间 QoS 1/2 的离线消息。对于要保证数据不丢的场景应设置为false。但代价是 broker 需要维护会话连接量大时内存占用会上升。completion-timeout消息发布或订阅时等待完成的超时时间单位毫秒。注意这是 Spring Integration 层面的阻塞时间上限不是网络超时。配置类的实现也很直白就是一个普通的ConfigurationProperties类Component ConfigurationProperties(prefix spring.mqtt) public class MqttProperties { private String brokerUrl; private String clientId; private String username; private String password; private String defaultTopic; private Integer completionTimeout 3000; private Boolean automaticReconnect true; // getter / setter 省略 }这样设计的好处是以后调整连接参数只改application.yml不用动 Java 代码。运维同事可以在不重新编译的情况下调整连接参数每次迭代都方便不少。4. 核心代码实现MQTT 配置类、订阅回调与消息发送链路4.1 配置类把 MQTT 客户端接入 Spring 容器所有的 MQTT 能力都通过一个Mqttv5PahoMessageDrivenChannelAdapter或MqttPahoMessageDrivenChannelAdapter注入 Spring 容器。这组类的设计思路是创建一个消息驱动的适配器让它持续监听 broker 上的消息一旦有消息进来就投递到指定的MessageChannel再由MessageHandler处理。先看连接工厂的创建Configuration public class MqttConfig { Bean public MqttPahoClientFactory mqttClientFactory() { DefaultMqttPahoClientFactory factory new DefaultMqttPahoClientFactory(); MqttConnectOptions options new MqttConnectOptions(); options.setServerURIs(new String[]{mqttProperties.getBrokerUrl()}); options.setAutomaticReconnect(mqttProperties.getAutomaticReconnect()); options.setCleanSession(true); options.setConnectionTimeout(10); // 如果需要用户名密码验证 options.setUserName(mqttProperties.getUsername()); options.setPassword(mqttProperties.getPassword().toCharArray()); factory.setConnectionOptions(options); return factory; } Bean public MqttPahoMessageDrivenChannelAdapter inboundAdapter() { MqttPahoMessageDrivenChannelAdapter adapter new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId(), mqttClientFactory(), mqttProperties.getDefaultTopic()); adapter.setCompletionTimeout(mqttProperties.getCompletionTimeout()); adapter.setQos(1); adapter.setOutputChannel(mqttInputChannel()); return adapter; } Bean public MessageChannel mqttInputChannel() { return new DirectChannel(); } }如果你用的是MqttPahoMessageDrivenChannelAdapterV3 版本上面的代码直接可用。网上很多旧教程用的就是 V3兼容性更好推荐先跑通再考虑 V5——V5 的适配器Mqttv5PahoMessageDrivenChannelAdapter对 MQTT 5.0 的属性支持更强但 API 稍有差异新手容易在 session expiry interval 这些参数上犯迷糊。开发初期用 V3 没毛病。DirectChannel是同步调用的意思是消息到达通道后当前的 adapter 线程会直接去执行 handler执行完再继续收下一条。如果业务处理逻辑耗时较长建议换成ExecutorChannel并用一个独立的线程池来处理消息避免阻塞 MQTT 消息接收线程。这里给一个带缓冲线程池的写法Bean public MessageChannel mqttInputChannel() { return new ExecutorChannel(Executors.newFixedThreadPool(8)); }像这种消息驱动的场景处理速度和接收速度不匹配是常态。只用DirectChannel时一旦业务处理慢了一半消息就会在 broker 侧积压消费者Lag越来越高。换线程池之后能缓冲一部分但也要注意别无限积压实际项目中最好配合监控来动态调整。4.2 消息回调与业务解耦把 MQTT 消息变成 Spring 事件适配器配置好之后下一步是消息的消费逻辑。Service 类里加一个方法用ServiceActivator注解标记Component public class MqttMessageReceiver { ServiceActivator(inputChannel mqttInputChannel) public void handleMessage(Message? message) throws MessagingException { String topic message.getHeaders().get(mqtt_receivedTopic, String.class); String payload new String((byte[]) message.getPayload(), StandardCharsets.UTF_8); System.out.println(收到主题: topic); System.out.println(消息内容: payload); } }这里有个细节需要特别说明Message.getPayload()的类型不一定是byte[]还可能已经是String取决于你配置的MessageConverter。如果遇到类型转换异常可以打印message.getPayload().getClass()来确认。这个看似小的点能帮你排除很多莫名的错误。更推荐的模式是回调里不做任何业务逻辑只负责把消息重新发布成 Spring 的ApplicationEvent这样业务模块不会依赖 MQTT 相关的类测试时直接 mock 事件发布即可Component public class MqttMessageReceiver { Autowired private ApplicationEventPublisher eventPublisher; ServiceActivator(inputChannel mqttInputChannel) public void handleMessage(Message? message) { String topic message.getHeaders().get(mqtt_receivedTopic, String.class); Object payload message.getPayload(); if (payload instanceof byte[]) { String content new String((byte[]) payload, StandardCharsets.UTF_8); eventPublisher.publishEvent(new MqttMessageEvent(this, topic, content)); } else { eventPublisher.publishEvent(new MqttMessageEvent(this, topic, payload.toString())); } } }事件类就用一个简单的 POJO 定义public class MqttMessageEvent extends ApplicationEvent { private final String topic; private final String payload; public MqttMessageEvent(Object source, String topic, String payload) { super(source); this.topic topic; this.payload payload; } // getter / setter 省略 }然后业务侧只需要Component public class DeviceDataListener { EventListener public void onMqttMessage(MqttMessageEvent event) { // 这里写你的业务逻辑比如保存数据库、调用其他服务 String topic event.getTopic(); String payload event.getPayload(); } }这套设计最直接的好处是——切断了 MQTT 协议层与业务层的耦合。以后哪怕你要把 MQTT 换成本地的消息队列业务代码都不用动只改回调里的转发逻辑就够了。另外事件监听机制天然支持多监听器比如一份数据进来可以同时触发持久化任务和实时推送任务扩展起来毫无压力。4.3 发布消息从普通字符串到动态主题订阅解决了接收问题发布则是反方向流程。Spring Integration 提供MqttPahoMessageHandler这本质上是一个MessageHandler实现它拿到消息后把数据发到 broker。标配代码如下Component public class MqttGateway { Autowired private MessageChannel mqttOutboundChannel; public void sendMessage(String topic, String payload) { mqttOutboundChannel.send(MessageBuilder.withPayload(payload) .setHeader(mqtt_topic, topic) .build()); } }对应的配置 BeanBean ServiceActivator(inputChannel mqttOutboundChannel) public MessageHandler mqttOutboundHandler() { MqttPahoMessageHandler handler new MqttPahoMessageHandler(springboot-mqtt-publisher, mqttClientFactory()); handler.setAsync(true); handler.setDefaultTopic(mqttProperties.getDefaultTopic()); return handler; }注意handler.setAsync(true)表示发布操作异步执行不会阻塞调用线程。同步发送setAsync(false)的话每条消息发送后要等 broker 确认高并发下吞吐量会明显下降。按照我的经验异步发送 合理的 QoS 策略是性能最好的组合。测试发布时可以用下面这段SpringBootTest class MqttPublishTest { Autowired private MqttGateway mqttGateway; Test void publishMessage() { mqttGateway.sendMessage(test/topic, hello springboot-mqtt); // 等异步发送完成再断言 Thread.sleep(1000); } }同时先用 mosquitto_sub 或 MQTTX 订阅该主题浏览器客户端或终端就会实时打出这条消息。这就构成最小可运行的闭环。如果消息没出现优先检查 4 个地方client-id 是否冲突、topic 是否写错、broker 地址是否通、用户名密码是否正确。我就遇到过 broker 地址从tcp://localhost:1883写成http://localhost:1883的错误排查了很久才发现是协议头的问题。broker 的默认端口是 1883改成了 http 的 8083 也会连不上这种低级错误一定要避免。5. 给 485 设备下发指令的场景MQTT 怎么和串口设备通信热搜词里有个非常典型的业务场景mqtt如何给485设备发指令、读取数据。这个需求在工业物联网领域极其常见。简单来说现场有大量通过 Modbus RTU 协议挂在 RS485 总线上的设备水表、电表、PLC、传感器等设备本身不支持 MQTT但采集终端DTU、边缘网关支持。架构上通常是边缘网关通过 RS485 总线读取设备数据再转换成 MQTT 消息上报到 broker。后端服务要下发指令比如远程开关阀门、修改设备参数时通过 MQTT 发送指令主题网关收到后转换成 Modbus RTU 指令通过串口发给目标设备。从服务端的角度MQTT 在这里承担的职责是透明传输。你需要做的并不复杂定义下行指令协议模板。比如主题cmd/{deviceId}/downlinkpayload 格式用 JSON 或者十六进制字符串。服务端发送指令时复用上面的MqttGateway。网关订阅cmd/#收到后解析出 deviceId查表找到对应的串口地址转成 Modbus 帧。实际项目中经常出现的问题是 QoS 和服务端应答的配合。如果下发指令而设备迟迟没有应答通常不是 MQTT 本身的问题而是网关和设备侧的总线链路问题。因此需要设计指令的请求-应答模式public class DownLinkCommand { private String requestId; private String deviceId; private String commandHex; private long timestamp; }服务端发完指令后在内存里缓存请求比如用ConcurrentHashMapString, CompletableFutureStringkey 用 requestId。设备上报数据时带上 requestId服务端在 MQTT 事件监听里把 future 完成掉从而实现同步转异步调用。这样上游调用方可以用同步方式等待抄表或指令执行结果但底层通信完全是异步的。这个模式就是物联网项目里指令下发 回调确认的标准解法。这里顺带提醒一下485 总线的通信距离长、抗干扰能力受布线影响大所以指令下发后设备没有应答的情况是常态。一定要给请求设置超时比如 3-5 秒没有收到设备响应就返回超时而不是无限阻塞线程。我见过有同事把 future 的等待时间设为 5 分钟结果同一时刻下发了大量指令服务端线程池直接被占满其他业务全部受影响。合理的超时是根据网关和设备的响应特点定的Modbus RTU 一般几百毫秒就能返回预留 2 秒足够。6. 探究消息持久化消息丢失、QoS 等级与 Session 机制的选择很多刚接触 MQTT 的同学会把 MQTT 类比成 Kafka 那样的消息队列认为发出去的消息一定能消费到。这是一个非常大的误区。MQTT 的定位是轻量级通信协议它不是为消息堆积而设计的。弄清楚 QoS 和 Session 这两个概念你才能设计出符合业务要求的消息可靠性模型。6.1 QoS 0/1/2 的实际表现差异QoS 0最多一次。消息发出后不确认可能丢失。适合传感器温度、湿度这类周期性上报的数据丢一次没关系下一轮还会来。QoS 1至少一次。消息发出后 broker 会回 PUBACK但客户端和 broker 之间可能出现消息重复。需要业务层做幂等。QoS 2只有一次。最严格的等级消息有完整的四步握手确认。性能开销较大台架测试中其吞吐量是 QoS 1 的 60% 左右不适合高并发场景。选择建议是设备上报数据一般用 QoS 1指令下发看业务要求如果不接受指令重复执行就考虑 QoS 2或者结合请求-应答去重。没有业务保障的情况下QoS 0 和 QoS 1 都可能造成消息重复或丢失这不是协议 bug而是 MQTT 的设计取舍。6.2 cleanSession 和 retained 到底控制什么cleanSessiontrue时broker 不会记录会话离线消息直接丢弃实时性很强但可靠性弱。cleanSessionfalse时broker 会保留会话和订阅关系客户端重新上线后可以收到离线期间积压的消息。代价是 broker 需要记录每个客户端的 session 状态连接数量多的时候内存会涨得厉害。还有一个容易混淆的概念是 retained message遗嘱消息。如果把主题的 retain 标志设为 truebroker 会为这个主题保留最后一条消息新订阅者连上来时立即收到这条消息即使是在它订阅之前发布的。这在告知新接入设备当前设备状态的场景下非常有用。比如设备上电后订阅device/{id}/status主题时立刻就能拿到 broker 里缓存的最后一条状态——而不是非要等设备下次主动上报。// 设置 retained 发布消息 MessageBuilder.withPayload(payload) .setHeader(mqtt_retained, true) .setHeader(mqtt_topic, topic) .build();6.3 落库策略消费方幂等是唯一可靠方案假设你决定用 QoS 1 接收设备上传的数据那么就必须在业务层想办法应对重复消息。最简单的方案是利用业务唯一键做幂等设备上报数据时都会带一个数据时间戳或消息序号这个序号在 broker 上不会有天然的全局去重能力所以要用 RedisSETNX或者数据库INSERT IGNORE来过滤重复。我在线上项目里还处理过一种更极端的情况同一台设备因为网络抖动在 broker 上创建了多个临时连接导致相同的消息被 broker 转发了两次甚至三次。如果消费端不做幂等数据库里会出现重复记录后面统计报表全部出错。所以幂等这种操作不属于优化而属于必须。7. 线上踩过的坑与排查思路7.1 掉线重连后订阅丢失Spring Integration 的适配器在内部管理订阅恢复。但如果你是在运行时动态增加过主题订阅adapter.addTopic(...)有些版本的重连机制恢复不了动态新增的主题等 broker 踢掉连接再重连之后数据就不再推送。排查时最明显的特征是日志里看不到任何报错但业务数据突然中断重启应用后又恢复。这类问题的根治办法尽量避免运行时动态订阅。所有需要的主题启动前就在配置文件里静态初始化。如果必须动态订阅重连后检查adapter.getTopic()的返回值和期望的订阅列表对比不一致就调用removeTopic()再addTopic()重建。另外检查 broker 的max_keepalive参数。如果 broker 设置的心跳上限比客户端的心跳间隔小客户端会被误判为死连接。两边参数务必对齐这个坑在跨团队协作时特别常见——设备组设置的心跳是 120 秒网关组设的周期是 30 秒跑到半夜连接全断。7.2 并发消费导致的消息顺序错乱同一个设备连续上报的两条数据如果消息被分发到两个不同的业务线程处理完成的先后顺序可能与发送顺序不一致。大部分场景下这不会有问题但碰到设备状态互斥这类依赖前后状态的业务结果会非常离谱。解决思路有两个方向使用ExecutorChannel时基于设备编号做路由保证同一个设备的所有消息进入同一个线程带 key 的线程池如 ThreadPoolExecutor 哈希取模。更稳妥的是在业务层设计状态机用设备上报的消息里的自增序号做顺序校验丢弃序号小于等于当前值的迟到消息。我自己在项目里选的是方案二因为方案一的哈希取模在设备数变化时会出现重排依然不可靠。序号校验虽然要写一点代码但换来的是逻辑层面的确定性。7.3 消息负载过大导致的内存溢出当 broker 或设备的消息频率过高时ExecutorChannel的线程池队列会持续积压。如果内存被撑爆应用直接 OOM现象和网络问题几乎一模一样毫无征兆。务必要做两件事给线程池的队列设置一个合理上限比如 10000和拒绝策略拒绝后由 MQTT 接收线程直接处理或丢弃。通过 Actuator 暴露队列大小指标。队列积压超过 80% 时报警这样就能在 OOM 前介入。下面是带队列上限的示例Bean(name mqttTaskExecutor) public Executor mqttExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); executor.setCorePoolSize(8); executor.setMaxPoolSize(16); executor.setQueueCapacity(10000); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.initialize(); return executor; }CallerRunsPolicy的语义是线程池满时由调用方线程直接执行任务。这里调用方线程是 MQTT 接收线程所以拒绝时相当于把压力回压到消息接收端让它们自然堆积在消息通道而不是无限塞进内存队列效果更温和。7.4 访问控制与安全认证如果 broker 直接暴露在内网或公网没做认证任何知道 IP 和端口的人都能订阅所有主题所有生产数据都会裸奔。这是非常致命的隐患。从最小成本的角度建议在 broker 配置用户名密码认证。敏感主题分层管理借助 ACL 限制比如device//status可订阅但device//cmd只能由固定客户端发布。管理后台和 broker 的 Web 控制台不要用默认端口暴露到公网。Mosquitto 的 ACL 配置示例大概长这样# mosquitto.conf allow_anonymous false password_file /etc/mosquitto/passwd acl_file /etc/mosquitto/acl # acl 文件 user device-reader topic read device//status user backend topic write device//cmd topic read device//status在这个配置下不同的服务账号权限是隔离的。订阅方拿不到下发主题的权限发布方也读不到其他设备的上报数据。这套最小权限模型在合规审计时也很好解释。8. 性能调优与数据可视化监控的接入思路当接入设备数量达到几千台时运行监控就不可逃避了。核心指标集中在 4 个连接数、订阅数、消息收发的吞吐量、消息积压时间。简单的方法是修改 application.yml暴露 Actuator MQTT 相关的指标。Spring Integration 提供了mqtt监控端点打开后可以观察 adapter 线程的工作状态MyBatis 或 JPA 侧也有数据源池指标整体都是通过 Prometheus 拉取。更细粒度的方向是引入 Micrometer 的自定义打点。把消息处理的耗时 T95/T99 记录到 Prometheus 直方图这样每次版本上线后性能有没有退化就一目了然。比如线上消息处理平均耗时 5ms压测时变成 20ms如果没打好点你可能根本不会发觉。之后配合 Grafana 看板你能看到消息时延线性的增长情况再主动做优化而不是等用户报故障。如果接入规模到了上万设备就需要考虑多 broker 集群方案比如 EMQX 的集群或 EMQX 的 NATS 桥接模式。但也要注意架构复杂度会陡增。设备连 A broker服务端连 B broker两边的消息路由需要配置好 bridge 规则。这个阶段要做的第一件事是明确业务跟 MQTT 之间的可靠性契约再去做集群规划。相同条件下使用单 EMQX 的分布式能力也可以管理十万级别的连接数不一定非得搞消息队列堆积中间件那套逻辑。9. MQTT 协议本身容易混淆的概念给刚入门的读者写这篇文章时想起来新手学习 MQTT 时其实有很多术语容易混淆这里挑几个重点快问快答。9.1 主题是有层级关系的/用于分级是单层通配符#是多层通配符。比如订阅sensor//temp能收到sensor/room1/temp和sensor/room2/temp订阅sensor/#能收到整个 sensor 分支下所有消息。注意#必须放在最后一位这是协议规范。$SYS/#这类$开头的主题是 broker 的系统主题存的是 broker 自身的运行数据比如客户端数量、消息流量。生产环境用$SYS/broker/clients/connected来做在线设备数监控是常见套路。9.2 遗嘱消息不是遗言遗嘱消息LWT是在设备异常断开时才发布的消息不是设备主动发的。比如设备正常在线时定期上报心跳正常状态如果设备断电或断网broker 在超时后自动发布一条遗嘱消息内容是预先设置好的设备离线。这套机制在设备在线状态监测里非常关键比等设备下次上报要实时得多。使用 Paho 配置遗嘱options.setWill(device/online/status, offline.getBytes(), 1, true);设备正常关闭时主动发送onlinefalse消息异常断电时broker 自动发遗嘱消息。两个路径不要重叠否则状态管理会很混乱。9.3 MQTT 5.0 多了什么如果 protocol 层选择 MQTT 5.0也就多了用户属性、消息过期时间、服务端断开原因码、共享订阅等能力。其中共享订阅$share/g1/topic能让多个客户端分摊同主题的消息流量对横向扩展很有意义。目前主流 Spring Integration 6.x 都能支持但需要显式指定使用 V5 的 adapter并且 broker 端也要打开 MQTT 5.0 支持。作为一个务实建议是先把自己的业务在 3.1.1 版本上玩熟再考虑 5.0 那部分新能力。协议版本越高能用的新特性越多但对基础设施的要求也越高。大多数业务的体量还轮不到 5.0 的高级特性来解决。10. 完整示例工程结构可直接复用的骨架这里给一个最小可运行的工程目录和关键代码文件方便你对照着手写一套自己的。技术栈是 Spring Boot 2.7.x spring-integration-mqtt5.1.x IDEA Maven。src/main/java/com/example/mqttdemo ├── config │ ├── MqttConfig.java │ └── MqttProperties.java ├── gateway │ └── MqttGateway.java ├── receiver │ └── MqttMessageReceiver.java ├── event │ └── MqttMessageEvent.java ├── listener │ └── DeviceDataListener.java └── MqttDemoApplication.java关键流程汇总如下直接作为项目实施清单用启动类保持最简骨架加SpringBootApplication即可。MqttProperties读取的配置项也齐全了记得加上EnableConfigurationProperties或ConfigurationPropertiesScan。适配器初始化后自动连接 broker。如果broker-url写错启动时不会报错但首次订阅或出消息时会抛连接异常本地开发时一定把 Mosquitto 或 MQTTX 打开确认连接成功。发布方只需要注入MqttGateway调用sendMessage(topic, payload)即可。接收方在DeviceDataListener里面用EventListener接收MqttMessageEvent填充自己的业务逻辑。这套结构跑通之后再往里面加 Spring Security、MyBatis、Redis 都不受影响。因为 MQTT 模块自成一体外部依赖都很常规。11. 我对 Spring Boot 整合 MQTT 的整体体会用 Spring Boot 做 MQTT 通信相比裸写 Paho 客户端最大的收益其实是把连接的建立与维护这个不变量从业务代码里剥离出来了。你的代码里不再出现while(true) { connect; }、重连线程这种东西统一交给适配器处理。连接稳定性、消息线程模型、重连处理在 Spring Integration 这套体系里都已经成熟了。踩过几次坑之后我的体会是MQTT 这种通信方式属于简单但不容易做好协议本身只有短短几百页但工程实践层面到处是细节。把 Broker 选型、QoS 选择、保持心跳、幂等处理、动态主题管理这些都列为项目检视清单启动前过一遍上线后再盯一阵基本不会出大事。如果想继续深挖建议下一步做这些扩展一是把 MQTT 消息接入到内存流式处理框架中做实时计算比如把温度、湿度传感器数据接入 Flink 这类流处理引擎另一个方向是研究配置中心与 MQTT 的结合让 Spring Boot 应用在运行期间动态更新订阅主题而不需要重启。每一条都足够做出一个独立项目来玩。希望这篇整理对你有实际帮助。