SpringBoot項目接入MQTT的詳細指南
一、引言
MQTT(Message Queuing Telemetry Transport)是一種輕量級的消息傳輸協(xié)議,特別適用于物聯(lián)網(wǎng)(IoT)場景,具有低帶寬、高延遲網(wǎng)絡(luò)環(huán)境下的優(yōu)勢。Spring Boot 作為流行的 Java 開發(fā)框架,能夠方便地與 MQTT 集成,實現(xiàn)高效的消息通信。本文將詳細介紹如何在 Spring Boot 項目中接入 MQTT。
二、環(huán)境準備
開發(fā)環(huán)境
- JDK 1.8 及以上版本
- Maven 3.x 或 Gradle
- Spring Boot 2.x 及以上版本
MQTT 服務(wù)器 可以選擇使用公共的 MQTT 服務(wù)器,如 HiveMQ 公共服務(wù)器(
tcp://broker.hivemq.com:1883),也可以自行搭建 Mosquitto 等 MQTT 服務(wù)器。
三、創(chuàng)建 Spring Boot 項目
可以使用 Spring Initializr(start.spring.io/)快速創(chuàng)建一個 Spring Boot 項目,添加以下依賴:
- Spring Web
- Spring for Apache Pulsar(因為 Pulsar 也支持 MQTT 協(xié)議,同時這里我們會使用其相關(guān)的 MQTT 依賴)
如果使用 Maven,pom.xml 中添加如下依賴:
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
</dependencies>
解釋
url:MQTT 服務(wù)器的地址和端口。client-id:客戶端的唯一標識。default-topic:默認訂閱的主題。username和password:如果 MQTT 服務(wù)器需要認證,則填寫相應(yīng)的用戶名和密碼。
五、創(chuàng)建 MQTT 配置類
創(chuàng)建一個配置類來配置 MQTT 連接和消息處理。
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
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.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
@Configuration
public class MqttConfig {
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[]{"${spring.mqtt.url}"});
options.setUserName("${spring.mqtt.username}");
options.setPassword("${spring.mqtt.password}".toCharArray());
factory.setConnectionOptions(options);
return factory;
}
@Bean
public MessageChannel mqttInputChannel() {
return new DirectChannel();
}
@Bean
public MqttPahoMessageDrivenChannelAdapter inbound() {
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter("${spring.mqtt.client-id}", mqttClientFactory(),
"${spring.mqtt.default-topic}");
adapter.setCompletionTimeout(5000);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
adapter.setOutputChannel(mqttInputChannel());
return adapter;
}
@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler handler() {
return message -> {
System.out.println("Received message: " + message.getPayload());
};
}
@Bean
public MessageChannel mqttOutputChannel() {
return new DirectChannel();
}
@Bean
@ServiceActivator(inputChannel = "mqttOutputChannel")
public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler =
new MqttPahoMessageHandler("${spring.mqtt.client-id}-publisher", mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic("${spring.mqtt.default-topic}");
return messageHandler;
}
}
解釋
mqttClientFactory:創(chuàng)建 MQTT 客戶端工廠,配置連接選項。mqttInputChannel和mqttOutputChannel:定義消息通道,用于接收和發(fā)送消息。inbound:創(chuàng)建 MQTT 消息驅(qū)動的通道適配器,用于訂閱主題并接收消息。handler:處理接收到的 MQTT 消息。mqttOutbound:創(chuàng)建 MQTT 消息處理程序,用于發(fā)布消息。
六、發(fā)送和接收 MQTT 消息
發(fā)送消息
創(chuàng)建一個服務(wù)類來發(fā)送 MQTT 消息。
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Service;
@Service
public class MqttMessageSender {
@Autowired
private MessageChannel mqttOutputChannel;
public void sendMessage(String message) {
mqttOutputChannel.send(new GenericMessage<>(message));
}
}
接收消息
在配置類中已經(jīng)定義了消息處理邏輯,當接收到消息時,會在 handler 方法中進行處理。
七、測試 MQTT 連接
創(chuàng)建一個控制器來測試 MQTT 消息的發(fā)送。
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class MqttController {
@Autowired
private MqttMessageSender mqttMessageSender;
@GetMapping("/send")
public String sendMessage(@RequestParam String message) {
mqttMessageSender.sendMessage(message);
return "Message sent: " + message;
}
}
啟動 Spring Boot 應(yīng)用程序,訪問 http://localhost:8080/send?message=Hello, MQTT! 即可發(fā)送 MQTT 消息。
八、總結(jié)
通過以上步驟,我們成功地在 Spring Boot 項目中接入了 MQTT,實現(xiàn)了消息的發(fā)送和接收。MQTT 作為一種輕量級的消息傳輸協(xié)議,與 Spring Boot 的集成可以幫助我們快速構(gòu)建高效、穩(wěn)定的物聯(lián)網(wǎng)消息通信系統(tǒng)。在實際應(yīng)用中,可以根據(jù)需求進一步擴展和優(yōu)化,如增加消息持久化、多主題訂閱等功能。
以上就是SpringBoot項目接入MQTT的詳細指南的詳細內(nèi)容,更多關(guān)于SpringBoot接入MQTT的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
Java后端實現(xiàn)異步編程的9種方式總結(jié)
我們?nèi)粘i_發(fā)的時候,經(jīng)常說到異步編程,比如說,在注冊接口,我們在用戶注冊成功時,用異步發(fā)送郵件通知用戶,那么實現(xiàn)異步編程一共有多少種方式呢,下面小編就來簡單講講吧2025-03-03
在Java中動態(tài)執(zhí)行字符串代碼的方法小結(jié)
在Java編程中,靜態(tài)編譯的特性通常不允許我們直接執(zhí)行運行時生成的代碼,然而,有時我們需要動態(tài)地生成并執(zhí)行代碼片段,本文將詳細介紹如何在Java中運行一段字符串代碼,并提供詳細的代碼案例和運行結(jié)果,需要的朋友可以參考下2024-08-08
spring-cloud入門之spring-cloud-config(配置中心)
這篇文章主要介紹了spring-cloud入門之spring-cloud-config(配置中心),小編覺得挺不錯的,現(xiàn)在分享給大家,也給大家做個參考。一起跟隨小編過來看看吧2018-01-01
如何在Spring?Boot微服務(wù)使用ValueOperations操作Redis集群String字符串
這篇文章主要介紹了在Spring?Boot微服務(wù)使用ValueOperations操作Redis集群String字符串類型數(shù)據(jù),本文通過實例代碼給大家介紹的非常詳細,對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2023-06-06

