在 Spring Boot 中使用 MQTT 通常會借助 Spring Integration 項目提供的 MQTT 支持。這使得 MQTT 的集成可以很好地融入 Spring 的消息驅動和企業集成模式。
以下是如何在 Spring Boot 中集成和使用 MQTT 的詳細步驟:
前提條件:
- MQTT Broker:需要一個正在運行的 MQTT Broker,例如 Mosquitto, EMQX, HiveMQ, RabbitMQ (with MQTT plugin), Apollo 等。確保 Broker 的地址和端口(默認通常是
tcp://localhost:1883
)。 - 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 等}*/
}
解釋:
-
MqttPahoClientFactory
:- 創建和配置底層的 Paho MQTT 客戶端。
MqttConnectOptions
用于設置 Broker URL、用戶名/密碼、Clean Session、Keep Alive、自動重連等。setCleanSession(true)
: 每次連接都是一個全新的會話,斷開后 Broker 不會保留訂閱信息和離線消息。setCleanSession(false)
: 持久會話,客戶端斷開重連后,Broker 會嘗試發送離線期間的 QoS 1 和 QoS 2 消息,并恢復訂閱。需要 Client ID 保持不變。
-
發布消息 (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。
-
訂閱消息 (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()
: 獲取消息內容。
-
@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:運行和測試
- 啟動 MQTT Broker (例如,使用 Docker 運行 Mosquitto):
docker run -it -p 1883:1883 -p 9001:9001 eclipse-mosquitto
- 運行 Spring Boot 應用。
- 測試發布:
- 如果創建了 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
。
- 如果創建了 REST Controller,可以通過 Postman 或 curl 發送 POST 請求:
- 測試訂閱:
- 當有消息發布到應用訂閱的主題 (例如
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"));
(需要相應的證書文件和密碼)
- 如果 Broker 使用 SSL/TLS,需要在
- 錯誤處理:
- Spring Integration 提供了錯誤通道 (
errorChannel
) 來處理消息傳遞過程中的異常。 - 可以監聽
MqttIntegrationEvent
(如MqttConnectionFailedEvent
) 來獲取連接狀態事件。
- Spring Integration 提供了錯誤通道 (
- Client ID 唯一性:再次強調,連接到同一個 Broker 的每個 MQTT 客戶端都必須有唯一的 Client ID。如果發布和訂閱邏輯在同一個應用實例中,并且為它們配置了相同的 Client ID 但使用了不同的
MqttPahoClientFactory
實例或適配器,Paho 庫內部可能會產生沖突或意外行為。建議為出站和入站適配器使用不同的 Client ID,或者共享一個MqttPahoClientFactory
實例(確保它能正確處理這種情況)。 - Converter (轉換器):
DefaultPahoMessageConverter
默認將String
轉換為byte[]
發送,接收時根據目標類型嘗試轉換。如果需要處理 JSON 或其他復雜類型,需要自定義MessageConverter
或在消息處理器中進行序列化/反序列化。