MQTT 在Spring Boot 中的使用

在 Spring Boot 中使用 MQTT 通常会借助 Spring Integration 项目提供的 MQTT 支持。这使得 MQTT 的集成可以很好地融入 Spring 的消息驱动和企业集成模式。

以下是如何在 Spring Boot 中集成和使用 MQTT 的详细步骤：

前提条件：

1. MQTT Broker：需要一个正在运行的 MQTT Broker，例如 Mosquitto, EMQX, HiveMQ, RabbitMQ (with MQTT plugin), Apollo 等。确保 Broker 的地址和端口（默认通常是 tcp://localhost:1883）。
2. Spring Boot 项目：一个基本的 Spring Boot 项目。

步骤 1：添加依赖

在你的 pom.xml (Maven) 或 build.gradle (Gradle) 文件中添加必要的依赖：

Maven (pom.xml):

<dependencies><!-- Spring Boot Starter for core functionality --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!-- Spring Boot Starter for Web (可选, 如果想通过 REST API 触发 MQTT 发布) --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- Spring Integration Core --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-integration</artifactId></dependency><!-- Spring Integration MQTT Support --><dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId><!-- Version managed by Spring Boot's BOM, or specify one --></dependency><!-- Eclipse Paho MQTT Client (Spring Integration MQTT uses this) --><dependency><groupId>org.eclipse.paho</groupId><artifactId>org.eclipse.paho.client.mqttv3</artifactId><version>1.2.5</version> <!-- Or a newer compatible version --></dependency>
</dependencies>

步骤 2：配置 MQTT 连接属性

在 src/main/resources/application.properties (或 application.yml) 中配置 MQTT Broker 的连接信息：

# MQTT Broker Configuration
mqtt.broker.url=tcp://localhost:1883
mqtt.client.id.publisher=springBootPublisher-unique # 客户端ID，对于每个连接必须唯一
mqtt.client.id.subscriber=springBootSubscriber-unique # 客户端ID，对于每个连接必须唯一
mqtt.default.topic=test/topic
mqtt.qos=1 # 默认的服务质量等级 (0, 1, or 2)# 可选：如果 Broker 需要认证
# mqtt.username=your_username
# mqtt.password=your_password

注意：Client ID 在 MQTT 中必须是唯一的。如果应用同时发布和订阅，可能需要为发布者和订阅者使用不同的 Client ID，或者使用一个 Client ID 但要确保 Paho 客户端实例的正确性。

步骤 3：创建 MQTT 配置类

创建一个 Java 配置类来定义 MQTT 相关的 Bean，如 ClientFactory、出站适配器（用于发布消息）和入站适配器（用于订阅消息）。

package com.example.mqttdemo.config;import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.handler.annotation.Header;@Configuration
@IntegrationComponentScan // 扫描 @MessagingGateway 等注解
public class MqttConfig {private static final Logger LOGGER = LoggerFactory.getLogger(MqttConfig.class);@Value("${mqtt.broker.url}")private String brokerUrl;@Value("${mqtt.client.id.publisher}")private String publisherClientId;@Value("${mqtt.client.id.subscriber}")private String subscriberClientId;@Value("${mqtt.default.topic}")private String defaultTopic;@Value("${mqtt.qos}")private int defaultQos;// 可选: 如果需要用户名密码认证// @Value("${mqtt.username}")// private String username;// @Value("${mqtt.password}")// private String password;// --- 通用 MQTT Client Factory ---@Beanpublic MqttPahoClientFactory mqttClientFactory() {DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();MqttConnectOptions options = new MqttConnectOptions();options.setServerURIs(new String[]{brokerUrl});// if (username != null && !username.isEmpty()) {//     options.setUserName(username);// }// if (password != null && !password.isEmpty()) {//     options.setPassword(password.toCharArray());// }options.setCleanSession(true); // 设置为 false 以启用持久会话和离线消息options.setAutomaticReconnect(true); // 启用自动重连options.setConnectionTimeout(10); // 连接超时时间 (秒)options.setKeepAliveInterval(20); // 心跳间隔 (秒)factory.setConnectionOptions(options);return factory;}// --- MQTT 消息发布 (Outbound) ---@Beanpublic MessageChannel mqttOutboundChannel() {return new DirectChannel(); // 或者 PublishSubscribeChannel}@Bean@ServiceActivator(inputChannel = "mqttOutboundChannel")public MessageHandler mqttOutbound(MqttPahoClientFactory clientFactory) {MqttPahoMessageHandler messageHandler =new MqttPahoMessageHandler(publisherClientId, clientFactory); // 使用独立的 Client IDmessageHandler.setAsync(true); // 推荐异步发送messageHandler.setDefaultTopic(defaultTopic); // 默认主题，可以被消息头覆盖messageHandler.setDefaultQos(defaultQos); // 默认QoS，可以被消息头覆盖// messageHandler.setDefaultRetained(false); // 默认是否保留消息return messageHandler;}// 定义一个网关接口，用于发送消息到 mqttOutboundChannel// Spring Integration 会自动实现这个接口@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")public interface MqttGateway {void sendToMqtt(String payload);void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);// 可以定义更多重载方法，例如发送 byte[]// void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, byte[] payload);}// --- MQTT 消息订阅 (Inbound) ---@Beanpublic MessageChannel mqttInputChannel() {return new DirectChannel();}@Beanpublic MqttPahoMessageDrivenChannelAdapter inboundAdapter(MqttPahoClientFactory clientFactory) {// 可以订阅单个主题，或多个主题 (字符串数组)// String[] topicsToSubscribe = {defaultTopic, "another/topic", "sensor/+/data"};MqttPahoMessageDrivenChannelAdapter adapter =new MqttPahoMessageDrivenChannelAdapter(subscriberClientId, clientFactory, defaultTopic); // 使用独立的 Client IDadapter.setCompletionTimeout(5000); // 等待消息发送完成的超时时间adapter.setConverter(new DefaultPahoMessageConverter()); // 消息转换器adapter.setQos(defaultQos); // 订阅时的QoSadapter.setOutputChannel(mqttInputChannel()); // 将接收到的消息发送到此 Channelreturn adapter;}// 消息处理器，处理从 mqttInputChannel 接收到的消息@ServiceActivator(inputChannel = "mqttInputChannel")public void handleIncomingMqttMessage(org.springframework.messaging.Message<String> message) {String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC, String.class);String payload = message.getPayload();Integer qos = message.getHeaders().get(MqttHeaders.RECEIVED_QOS, Integer.class);Boolean retained = message.getHeaders().get(MqttHeaders.RETAINED, Boolean.class);LOGGER.info("Received MQTT Message - Topic: [{}], QoS: [{}], Retained: [{}], Payload: [{}]",topic, qos, retained, payload);// 在这里处理你的业务逻辑}// 可选: 监听 MQTT 事件 (连接成功，连接丢失等)/*@EventListenerpublic void handleMqttEvents(MqttIntegrationEvent event) {LOGGER.info("MQTT Event: {}", event);if (event instanceof MqttConnectionFailedEvent) {MqttConnectionFailedEvent failedEvent = (MqttConnectionFailedEvent) event;LOGGER.error("MQTT Connection Failed!", failedEvent.getCause());} else if (event instanceof MqttSubscribedEvent) {MqttSubscribedEvent subscribedEvent = (MqttSubscribedEvent) event;LOGGER.info("MQTT Subscribed to: {}", subscribedEvent.getMessage());}//还有 MqttMessageSentEvent, MqttMessageDeliveredEvent 等}*/
}

解释：

1. MqttPahoClientFactory:

   • 创建和配置底层的 Paho MQTT 客户端。
   • MqttConnectOptions 用于设置 Broker URL、用户名/密码、Clean Session、Keep Alive、自动重连等。
   • setCleanSession(true): 每次连接都是一个全新的会话，断开后 Broker 不会保留订阅信息和离线消息。
   • setCleanSession(false): 持久会话，客户端断开重连后，Broker 会尝试发送离线期间的 QoS 1 和 QoS 2 消息，并恢复订阅。需要 Client ID 保持不变。

2. 发布消息 (Outbound):

   • mqttOutboundChannel: 一个 MessageChannel，作为消息发布的入口。
   • MqttPahoMessageHandler (mqttOutbound bean): 这是一个 MessageHandler，它监听 mqttOutboundChannel，并将接收到的消息通过 MQTT 发送出去。
     • 需要一个 clientId 和 MqttPahoClientFactory。
     • setAsync(true): 异步发送消息，不会阻塞当前线程。
     • setDefaultTopic() 和 setDefaultQos(): 如果消息本身没有指定主题或QoS，则使用这些默认值。
   • MqttGateway: 一个接口，使用 @MessagingGateway 注解。Spring Integration 会自动为其生成实现。通过调用这个接口的方法，可以方便的将消息发送到 defaultRequestChannel (即 mqttOutboundChannel)。
     • @Header(MqttHeaders.TOPIC) 和 @Header(MqttHeaders.QOS) 允许在发送时动态指定主题和QoS。

3. 订阅消息 (Inbound):

   • mqttInputChannel: 一个 MessageChannel，入站适配器会将从 MQTT Broker 收到的消息发送到这个 Channel。
   • MqttPahoMessageDrivenChannelAdapter (inboundAdapter bean): 这是一个消息驱动的通道适配器，它连接到 MQTT Broker，订阅指定的主题，并将收到的消息推送到 outputChannel (即 mqttInputChannel)。
     • 需要一个 clientId、MqttPahoClientFactory 和要订阅的主题（可以是一个或多个，支持通配符）。
     • setConverter(): 用于将 MQTT 的 byte[] 负载转换为期望的类型（例如 String）。DefaultPahoMessageConverter 可以处理 String 和 byte[]。
   • handleIncomingMqttMessage 方法: 使用 @ServiceActivator(inputChannel = "mqttInputChannel") 注解，监听 mqttInputChannel。当有消息到达时，此方法会被调用。
     • message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC): 获取消息来源的主题。
     • message.getPayload(): 获取消息内容。

4. @IntegrationComponentScan: 确保 Spring Integration 扫描并处理 @MessagingGateway 等注解。

步骤 4：使用 MQTT Gateway 发布消息

你可以注入 MqttGateway 到你的 Service 或 Controller 中来发送消息。

示例 Service:

package com.example.mqttdemo.service;import com.example.mqttdemo.config.MqttConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;@Service
public class MqttPublishService {private static final Logger LOGGER = LoggerFactory.getLogger(MqttPublishService.class);@Autowiredprivate MqttConfig.MqttGateway mqttGateway;public void publishMessage(String topic, String payload) {try {LOGGER.info("Publishing MQTT message - Topic: [{}], Payload: [{}]", topic, payload);mqttGateway.sendToMqtt(topic, payload);} catch (Exception e) {LOGGER.error("Error publishing MQTT message to topic {}: {}", topic, e.getMessage(), e);}}public void publishMessageWithQos(String topic, String payload, int qos) {try {LOGGER.info("Publishing MQTT message - Topic: [{}], QoS: [{}], Payload: [{}]", topic, qos, payload);mqttGateway.sendToMqtt(topic, qos, payload);} catch (Exception e) {LOGGER.error("Error publishing MQTT message to topic {} with QoS {}: {}", topic, qos, e.getMessage(), e);}}public void publishToDefaultTopic(String payload) {try {LOGGER.info("Publishing MQTT message to default topic, Payload: [{}]", payload);mqttGateway.sendToMqtt(payload); // 将使用 MqttPahoMessageHandler 中配置的默认主题和QoS} catch (Exception e) {LOGGER.error("Error publishing MQTT message to default topic: {}", e.getMessage(), e);}}
}

示例 REST Controller (可选):

package com.example.mqttdemo.controller;import com.example.mqttdemo.service.MqttPublishService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;@RestController
@RequestMapping("/api/mqtt")
public class MqttController {@Autowiredprivate MqttPublishService mqttPublishService;@PostMapping("/publish")public String publishMessage(@RequestParam String topic, @RequestBody String payload) {mqttPublishService.publishMessage(topic, payload);return "Message published to topic: " + topic;}@PostMapping("/publish-default")public String publishToDefault(@RequestBody String payload) {mqttPublishService.publishToDefaultTopic(payload);return "Message published to default topic.";}@PostMapping("/publish-qos")public String publishMessageWithQos(@RequestParam String topic,@RequestParam int qos,@RequestBody String payload) {mqttPublishService.publishMessageWithQos(topic, payload, qos);return "Message published to topic: " + topic + " with QoS: " + qos;}
}

步骤 5：运行和测试

1. 启动 MQTT Broker (例如，使用 Docker 运行 Mosquitto):

   docker run -it -p 1883:1883 -p 9001:9001 eclipse-mosquitto
   

2. 运行 Spring Boot 应用。
3. 测试发布：
   • 如果创建了 REST Controller，可以通过 Postman 或 curl 发送 POST 请求：
     POST http://localhost:8080/api/mqtt/publish?topic=my/custom/topic
     Body (raw, text/plain): Hello from Spring Boot MQTT!
   • 或者在应用启动时通过 CommandLineRunner 调用 MqttPublishService。
4. 测试订阅：
   • 当有消息发布到应用订阅的主题 (例如 test/topic 或在 inboundAdapter 中配置的其他主题) 时，handleIncomingMqttMessage 方法会被调用，在控制台能看到日志输出。
   • 也可以使用 MQTT 客户端工具 (如 MQTTX, MQTT Explorer) 连接到同一个 Broker 并发布消息到被订阅的主题。

高级主题和注意事项：

• QoS (服务质量等级)：
  • QoS 0 (At most once): 最多一次，消息可能丢失。
  • QoS 1 (At least once): 至少一次，消息可能重复。
  • QoS 2 (Exactly once): 精确一次，最可靠但开销最大。
  • 可以在 MqttPahoMessageHandler 和 MqttPahoMessageDrivenChannelAdapter 中设置默认 QoS，也可以在发送消息时通过 MqttHeaders.QOS 动态指定。
• Retained Messages (保留消息)：
  • 发布者可以将消息标记为 “retained”。Broker 会存储该主题下最新的保留消息。当新客户端订阅该主题时，会立即收到这条保留消息。
  • 在 MqttPahoMessageHandler 中设置 setDefaultRetained(true) 或通过消息头 MqttHeaders.RETAINED。
• Last Will and Testament (LWT - 遗嘱消息)：
  • 在 MqttConnectOptions 中配置。如果客户端异常断开，Broker 会发布这个预设的遗嘱消息到指定主题。
  • options.setWill("client/status", "offline".getBytes(), 1, false);
• SSL/TLS 加密：
  • 如果 Broker 使用 SSL/TLS，需要在 MqttConnectOptions 中配置 SSL 属性，并将 Broker URL 改为 ssl://your-broker-address:8883。
  • options.setSocketFactory(SslUtil.getSocketFactory("ca.crt", "client.crt", "client.key", "password")); (需要相应的证书文件和密码)
• 错误处理：
  • Spring Integration 提供了错误通道 (errorChannel) 来处理消息传递过程中的异常。
  • 可以监听 MqttIntegrationEvent (如 MqttConnectionFailedEvent) 来获取连接状态事件。
• Client ID 唯一性：再次强调，连接到同一个 Broker 的每个 MQTT 客户端都必须有唯一的 Client ID。如果发布和订阅逻辑在同一个应用实例中，并且为它们配置了相同的 Client ID 但使用了不同的 MqttPahoClientFactory 实例或适配器，Paho 库内部可能会产生冲突或意外行为。建议为出站和入站适配器使用不同的 Client ID，或者共享一个 MqttPahoClientFactory 实例（确保它能正确处理这种情况）。
• Converter (转换器): DefaultPahoMessageConverter 默认将 String 转换为 byte[] 发送，接收时根据目标类型尝试转换。如果需要处理 JSON 或其他复杂类型，需要自定义 MessageConverter 或在消息处理器中进行序列化/反序列化。