Springboot對接mqtt的項目實(shí)踐
更新時間：2026年02月10日 09:50:27 作者：IT界Tony哥
在SpringBoot中對接MQTT協(xié)議,可以使用Eclipse Paho客戶端和Spring Integration MQTT模塊,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
在Spring Boot中對接MQTT協(xié)議，可以使用Eclipse Paho客戶端和Spring Integration MQTT模塊。以下是詳細(xì)實(shí)現(xiàn)步驟：
1. 添加依賴
<dependencies>
<!-- Spring Boot Starter -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<!-- Spring Integration MQTT -->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
<!-- Eclipse Paho MQTT Client -->
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>
</dependencies>
2. 配置MQTT連接參數(shù)
# application.yml mqtt: broker-url: tcp://localhost:1883 username: admin password: password client-id: spring-boot-client default-topic: test/topic timeout: 30 keepalive: 60 completion-timeout: 30000
3. MQTT配置類
@Configuration
@EnableConfigurationProperties(MqttProperties.class)
public class MqttConfig {
@Autowired
private MqttProperties mqttProperties;
// MQTT連接配置
@Bean
public MqttConnectOptions mqttConnectOptions() {
MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[]{mqttProperties.getBrokerUrl()});
options.setUserName(mqttProperties.getUsername());
options.setPassword(mqttProperties.getPassword().toCharArray());
options.setConnectionTimeout(mqttProperties.getTimeout());
options.setKeepAliveInterval(mqttProperties.getKeepalive());
options.setAutomaticReconnect(true);
options.setCleanSession(true);
return options;
}
// MQTT客戶端工廠
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setConnectionOptions(mqttConnectOptions());
return factory;
}
// 出站消息通道（用于發(fā)送消息）
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler =
new MqttPahoMessageHandler(mqttProperties.getClientId() + "-producer", mqttClientFactory());
messageHandler.setAsync(true);
messageHandler.setDefaultTopic(mqttProperties.getDefaultTopic());
return messageHandler;
}
// 出站通道
@Bean
public MessageChannel mqttOutboundChannel() {
return new DirectChannel();
}
// 入站消息適配器（用于接收消息）
@Bean
public MessageProducer inbound() {
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId() + "-consumer",
mqttClientFactory(), mqttProperties.getDefaultTopic());
adapter.setCompletionTimeout(mqttProperties.getCompletionTimeout());
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
adapter.setOutputChannel(mqttInboundChannel());
return adapter;
}
// 入站通道
@Bean
public MessageChannel mqttInboundChannel() {
return new DirectChannel();
}
// 入站消息處理器
@Bean
@ServiceActivator(inputChannel = "mqttInboundChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
String topic = (String) message.getHeaders().get("mqtt_receivedTopic");
String payload = (String) message.getPayload();
System.out.println("Received message from topic: " + topic + ", payload: " + payload);
// 處理接收到的消息
processMessage(topic, payload);
}
};
}
}
4. 配置屬性類
@ConfigurationProperties(prefix = "mqtt")
@Component
@Data
public class MqttProperties {
private String brokerUrl;
private String username;
private String password;
private String clientId;
private String defaultTopic;
private int timeout;
private int keepalive;
private int completionTimeout;
}
5. MQTT服務(wù)類
@Service
public class MqttService {
@Autowired
private MessageChannel mqttOutboundChannel;
// 發(fā)送消息到指定主題
public void sendMessage(String topic, String message) {
mqttOutboundChannel.send(MessageBuilder.withPayload(message)
.setHeader("mqtt_topic", topic)
.build());
}
// 發(fā)送消息到默認(rèn)主題
public void sendMessage(String message) {
mqttOutboundChannel.send(MessageBuilder.withPayload(message).build());
}
// 發(fā)送帶QoS的消息
public void sendMessage(String topic, String message, int qos) {
mqttOutboundChannel.send(MessageBuilder.withPayload(message)
.setHeader("mqtt_topic", topic)
.setHeader("mqtt_qos", qos)
.build());
}
}
6. 消息處理器
@Component
public class MqttMessageProcessor {
private static final Logger logger = LoggerFactory.getLogger(MqttMessageProcessor.class);
public void processMessage(String topic, String payload) {
logger.info("Processing MQTT message - Topic: {}, Payload: {}", topic, payload);
// 根據(jù)不同的主題進(jìn)行不同的處理
switch (topic) {
case "test/topic":
handleTestTopic(payload);
break;
case "sensor/data":
handleSensorData(payload);
break;
default:
handleDefaultMessage(topic, payload);
}
}
private void handleTestTopic(String payload) {
logger.info("處理測試主題消息: {}", payload);
// 具體的業(yè)務(wù)邏輯
}
private void handleSensorData(String payload) {
logger.info("處理傳感器數(shù)據(jù): {}", payload);
try {
// 解析JSON數(shù)據(jù)等操作
// ObjectMapper mapper = new ObjectMapper();
// SensorData data = mapper.readValue(payload, SensorData.class);
} catch (Exception e) {
logger.error("解析傳感器數(shù)據(jù)失敗", e);
}
}
private void handleDefaultMessage(String topic, String payload) {
logger.info("處理默認(rèn)消息 - Topic: {}, Payload: {}", topic, payload);
}
}
7. 控制器示例
@RestController
@RequestMapping("/mqtt")
public class MqttController {
@Autowired
private MqttService mqttService;
@PostMapping("/publish")
public ResponseEntity<String> publishMessage(@RequestParam String topic,
@RequestParam String message) {
try {
mqttService.sendMessage(topic, message);
return ResponseEntity.ok("Message published successfully");
} catch (Exception e) {
return ResponseEntity.status(500).body("Failed to publish message: " + e.getMessage());
}
}
@PostMapping("/publish/default")
public ResponseEntity<String> publishToDefaultTopic(@RequestParam String message) {
try {
mqttService.sendMessage(message);
return ResponseEntity.ok("Message published to default topic");
} catch (Exception e) {
return ResponseEntity.status(500).body("Failed to publish message: " + e.getMessage());
}
}
}
8. 主應(yīng)用類
@SpringBootApplication
@EnableConfigurationProperties
public class MqttApplication {
public static void main(String[] args) {
SpringApplication.run(MqttApplication.class, args);
}
}
9. 測試MQTT服務(wù)
可以使用MQTT.fx或其他MQTT客戶端工具進(jìn)行測試：
- 啟動Spring Boot應(yīng)用
- 使用MQTT客戶端訂閱主題
test/topic - 調(diào)用API發(fā)送消息：
curl -X POST "http://localhost:8080/mqtt/publish?topic=test/topic&message=Hello MQTT"
主要特性
- 自動重連: 配置了自動重連機(jī)制
- QoS支持: 支持不同的服務(wù)質(zhì)量等級
- 多主題訂閱: 可以訂閱多個主題
- 異步處理: 消息發(fā)送支持異步模式
- 配置靈活: 通過配置文件管理連接參數(shù)
這樣你就實(shí)現(xiàn)了一個完整的Spring Boot MQTT集成方案，可以方便地進(jìn)行消息的發(fā)布和訂閱。
到此這篇關(guān)于Springboot對接mqtt的項目實(shí)踐的文章就介紹到這了,更多相關(guān)Springboot對接mqtt內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
相關(guān)文章
這篇文章主要介紹了SpringBoot @NotBlank錯誤的解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-08-08
這篇文章主要介紹了分享JVM 的四種引用方式，文章圍繞主題展開詳細(xì)的內(nèi)容介紹，具有一定的參考價值，需要的小伙伴可以參考一下2022-07-07
這篇文章主要介紹了java中帶參數(shù)的try(){}語法含義詳解，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-02-02
這篇文章主要給大家介紹了關(guān)于Spring Cloud Zuul過濾器的相關(guān)資料,通過閱讀本文您將了解：Zuul過濾器類型與請求生命周期、如何編寫Zuul過濾器、如何禁用Zuul過濾器和Spring Cloud為Zuul編寫的過濾器及其功能，需要的朋友可以參考下。2017-02-02
spring AOP的After增強(qiáng)實(shí)現(xiàn)方法實(shí)例分析
這篇文章主要介紹了spring AOP的After增強(qiáng)實(shí)現(xiàn)方法,結(jié)合實(shí)例形式分析了spring面向切面AOP的After增強(qiáng)實(shí)現(xiàn)步驟與相關(guān)操作技巧,需要的朋友可以參考下2020-01-01
這篇文章主要為大家詳細(xì)介紹了java實(shí)現(xiàn)單鏈表中的增刪改，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-05-05
Java制表符與空格的轉(zhuǎn)換之EnTab和DeTab的使用
本文將深入探討如何使用Java的EnTab和DeTab類實(shí)現(xiàn)制表符與空格的智能轉(zhuǎn)換,并分析其在實(shí)際開發(fā)中的應(yīng)用,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2025-04-04
Java數(shù)據(jù)結(jié)構(gòu)順序表用法詳解
順序表是計算機(jī)內(nèi)存中以數(shù)組的形式保存的線性表，線性表的順序存儲是指用一組地址連續(xù)的存儲單元依次存儲線性表中的各個元素、使得線性表中在邏輯結(jié)構(gòu)上相鄰的數(shù)據(jù)元素存儲在相鄰的物理存儲單元中，即通過數(shù)據(jù)元素物理存儲的相鄰關(guān)系來反映數(shù)據(jù)元素之間邏輯上的相鄰關(guān)系2021-10-10
本文主要介紹了servlet創(chuàng)建web后端程序的示例代碼，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-06-06
這篇文章主要介紹了Java工程如何打印程序日志過程解析,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-10-10