物联网项目里最常见的需求就是设备与服务端之间的消息互通。传感器每隔几秒上报一次温度数据，服务端收到后写入数据库；用户在手机上点一下开关，指令要实时推送到智能硬件上。这类场景如果直接用HTTP接口轮询，不仅浪费带宽，实时性也难以保证。MQTT作为一种基于发布订阅模式的轻量级消息协议，专门为低带宽、不稳定网络环境设计，天然适合物联网通信。本文将以Spring Boot为基础，完整演示如何整合MQTT实现设备消息的收发。
一、理解MQTT协议的核心机制
MQTT的通信模型是发布订阅模式，而不是传统的请求响应模式。发送方（发布者）不直接把消息发给接收方，而是把消息发送到某个主题（Topic）上；接收方（订阅者）向Broker订阅自己关心的主题，只要有新消息到达该主题，Broker就会把消息推送给所有订阅者。这种解耦方式让设备和服务端互相不需要知道对方的地址，扩容时只需要增加订阅者即可。
主题采用类似文件路径的层级结构，比如device/sensor001/temperature表示编号001的传感器的温度数据。订阅时可以使用通配符：+匹配单层，例如device/+/temperature能匹配所有设备的温度消息；#匹配多层，例如device/#能匹配device下所有子主题的消息。合理设计主题结构，可以让后端用最少的订阅规则覆盖全部设备。
QoS（服务质量）等级是MQTT的另一个重点。QoS 0表示最多发送一次，消息可能丢失，适合高频且允许丢帧的数据；QoS 1表示至少发送一次，可能重复，需要消费端做幂等处理；QoS 2表示恰好一次，开销最大，适合计费类、控制类的关键指令。设备上报温度用QoS 0或1就够了，而下发开锁、支付确认这类指令建议用QoS 2。
MQTT还支持遗嘱消息（Will Message）机制。客户端连接时可以预先设置一条遗嘱消息，一旦客户端异常掉线（没有正常发送断开包），Broker会自动向指定主题发布这条遗嘱。服务端订阅设备遗嘱主题，就能感知设备意外离线的事件，这是实现设备在线状态监控的关键手段。
二、在Spring Boot中整合Eclipse Paho客户端
Java生态里最常用的MQTT客户端是Eclipse Paho，Spring Boot整合它非常简单。首先在pom.xml中引入依赖，然后编写一个配置类，在应用启动时创建MQTT连接。
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId;gt;org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>上面的写法要注意artifactId标签必须是完整的<artifactId>org.eclipse.paho.client.mqttv3</artifactId>。接下来编写配置类，把MqttClient注册为Spring Bean，方便在业务代码中随时注入使用。
@Configuration
public class MqttConfig {
@Bean
public MqttClient mqttClient() throws MqttException {
// 内存持久化，重启后清空；需要消息可靠落地可换成 MqttDefaultFilePersistence
MqttClient client = new MqttClient(
"tcp://127.0.0.1:1883",
"springboot-server-client",
new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("admin");
options.setPassword("123456".toCharArray());
options.setCleanSession(false); // 保留会话，断线期间的消息不会被丢弃
options.setAutomaticReconnect(true); // 自动重连
options.setConnectionTimeout(10);
options.setKeepAliveInterval(30); // 心跳间隔，单位秒
// 设置遗嘱：服务端异常掉线时通知其他系统
options.setWill("server/status", "offline".getBytes(), 1, false);
client.connect(options);
return client;
}
}这里有几个参数值得说明。cleanSession设为false后，Broker会为该客户端保留会话状态，断线期间的QoS 1和QoS 2消息会在重连后补发。keepAliveInterval是心跳周期，客户端会定期发送PING报文维持连接，如果Broker在一个半周期内没收到任何报文，就会判定客户端掉线并发布遗嘱消息。生产环境建议开启setAutomaticReconnect，避免网络抖动导致连接永久失效。
需要注意，Paho的MqttClient本身不是线程安全的发布瓶颈，高并发场景下建议改用MqttAsyncClient，它的所有操作都是异步的，不会阻塞业务线程。另外连接ID（clientId）在同一个Broker内必须唯一，如果两个客户端用同一个ID连接，旧连接会被强制踢下线，这在多实例部署时尤其要小心。
三、实现消息发布与订阅处理
客户端连接好之后，先实现消息发布。写一个简单的Service，封装发布方法，把异常处理也一并做好。
@Service
public class MqttPublisher {
@Autowired
private MqttClient mqttClient;
public void publish(String topic, String payload, int qos) {
try {
MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
message.setQos(qos);
mqttClient.publish(topic, message);
} catch (MqttException e) {
// 记录日志并考虑重试或落库补偿
log.error("MQTT消息发布失败, topic={}", topic, e);
throw new RuntimeException("MQTT publish failed", e);
}
}
}订阅消息需要实现回调接口。MqttCallbackExtended比基础的MqttCallback多了一个connectComplete方法，重连成功后可以在这里重新订阅主题，因为某些Broker在会话丢失后会取消订阅关系。
@Component
public class DeviceMessageListener implements MqttCallbackExtended {
@Autowired
private MqttClient mqttClient;
@PostConstruct
public void subscribe() throws MqttException {
mqttClient.setCallback(this);
// 订阅所有设备的上行数据
mqttClient.subscribe("device/+/upstream", 1);
// 订阅设备遗嘱主题，感知设备掉线
mqttClient.subscribe("device/+/will", 1);
}
@Override
public void messageArrived(String topic, MqttMessage message) {
String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
if (topic.endsWith("/upstream")) {
// 解析并处理设备上报数据，注意QoS 1下消息可能重复，需做幂等
System.out.println("收到设备数据: " + payload);
} else if (topic.endsWith("/will")) {
String deviceId = topic.split("/")[1];
System.out.println("设备离线: " + deviceId);
}
}
@Override
public void connectionLost(Throwable cause) {
System.out.println("MQTT连接断开: " + cause.getMessage());
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) { }
@Override
public void connectComplete(boolean reconnect, String serverURI) {
if (reconnect) {
try {
// 重连后重新订阅，防止会话丢失导致订阅失效
mqttClient.subscribe(new String[]{"device/+/upstream", "device/+/will"});
} catch (MqttException e) {
e.printStackTrace();
}
}
}
}处理上行消息时有几个坑要提前防住。第一，messageArrived是在Paho内部线程中回调的，如果消息处理逻辑很重（比如涉及数据库批量写入），应该把消息丢进线程池或消息队列异步处理，否则会阻塞后续消息的接收，造成堆积。第二，QoS 1的消息可能重复到达，业务层必须设计幂等键，比如用消息ID或设备上报的时间戳去重。第三，设备消息量大的情况下，单台Spring Boot实例的订阅吞吐可能不够，可以按topic通配符拆分，让不同实例订阅不同范围的设备。
四、Broker选型与生产部署建议
Broker是MQTT架构的核心，常见的选择有EMQX、Mosquitto和HiveMQ。Mosquitto轻量简单，适合小规模部署和开发测试；EMQX支持分布式集群、规则引擎和消息桥接，单节点能承载百万级连接，是国内物联网项目的主流选择。生产环境建议给Broker配置TLS加密传输，设备端使用证书认证，避免明文传输被嗅探或伪造设备接入。
架构上还有一点值得考虑：不要让业务服务直接消费所有设备消息。更稳妥的做法是Spring Boot服务只做协议接入层，把收到的消息转发到Kafka或RabbitMQ，再由下游的数据清洗、告警、存储服务分别消费。这样MQTT层与业务层彻底解耦，业务服务重启或升级时不会造成设备消息丢失，整个系统的可扩展性也会好很多。
总结一下，Spring Boot整合MQTT的关键点在于：理解发布订阅模型与主题设计，正确配置QoS等级和遗嘱消息，处理好断线重连与幂等消费，再配合可靠的Broker完成部署。把这些环节都做到位，一套稳定高效的物联网消息通道就搭建完成了。
Spring BootMQTT物联网修改时间：2026-09-09 18:55:12