MQTT SpringBoot整合實戰(zhàn)教程

更新時間：2025年05月28日 10:54:11 作者：菜逼の世界

這篇文章主要介紹了MQTT SpringBoot整合實戰(zhàn)教程,本文通過實例代碼給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友參考下吧

MQTT-SpringBoot

創(chuàng)建簡單 SpringBoot 項目

導入必須依賴

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.study</groupId>
    <artifactId>MqttDemo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>SpringBootMqttDemo</name>
    <description>SpringBootMqttDemo</description>
    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.6.13</spring-boot.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- spring boot項目web開發(fā)的起步依賴 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- spring boot項目集成消息中間件基礎依賴 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-integration</artifactId>
        </dependency>
        <!-- spring boot項目和mqtt客戶端集成起步依賴 -->
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
            <version>5.4.3</version>
        </dependency>
        <!-- lombok依賴 -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <!-- fastjson依賴 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.83</version>
        </dependency>
    </dependencies>
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>${spring-boot.version}</version>
                <configuration>
                    <mainClass>com.study.mqtt.demo.MqttDemoApplication</mainClass>
                    <skip>true</skip>
                </configuration>
                <executions>
                    <execution>
                        <id>repackage</id>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

增加MQTT相關配置

application.yml

spring:
  mqtt:
    # mqtt 服務器地址
    url: tcp://192.168.40.128:1883
    # 訂閱客戶端ID
    subClientId: sub_client_id_1
    # 訂閱主題
    subTopic: lq/iot/demo/
    # 發(fā)布客戶端ID
    pubClientId: pub_client_id_1
    # 用戶名
    username: admin
    # 密碼
    password: admin123456

編寫對應Java類

配置類

MqttConfig.java

package com.study.mqtt.demo.domain;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
@Data
@ConfigurationProperties(prefix = "spring.mqtt")
public class MqttConfig {
    private String username;
    private String password;
    private String url;
    private String subClientId ;
    private String subTopic ;
    private String pubClientId ;
}

啟動類增加開啟配置

MqttDemoApplication.java

package com.study.mqtt.demo;
import com.study.mqtt.demo.domain.MqttConfig;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
@SpringBootApplication
@EnableConfigurationProperties(value = MqttConfig.class)
public class MqttDemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(MqttDemoApplication.class, args);
    }
}

創(chuàng)建MQTT連接工廠類

MqttFactory.java

package com.study.mqtt.demo.factory;
import com.study.mqtt.demo.domain.MqttConfig;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
@Configuration
public class MqttFactory {
    @Autowired
    private MqttConfig mqttConfig;
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        // 創(chuàng)建客戶端工廠
        DefaultMqttPahoClientFactory  factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(mqttConfig.getUsername());
        options.setPassword(mqttConfig.getPassword().toCharArray());
        options.setServerURIs(new String[]{mqttConfig.getUrl()});
        options.setCleanSession(true);
        factory.setConnectionOptions(options);
        return factory;
    }
}

接收消息處理類

ReceiveMsgHandler.java

package com.study.mqtt.demo.handler;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;
@Component
public class ReceiveMsgHandler implements MessageHandler {
    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        System.out.println("接收到消息對象：" + message);
        // 消息內(nèi)容
        Object payload = message.getPayload();
        MessageHeaders headers = message.getHeaders();
        Object mqttReceivedTopic = headers.get("mqtt_receivedTopic");
        System.out.println("接收的消息主題：" + mqttReceivedTopic);
        System.out.println("接收的消息內(nèi)容：" + payload);
    }
}

接收消息配置類

MqttInboundConfig.java

package com.study.mqtt.demo.inbound;
import com.study.mqtt.demo.domain.MqttConfig;
import com.study.mqtt.demo.factory.MqttFactory;
import com.study.mqtt.demo.handler.ReceiveMsgHandler;
import org.springframework.beans.factory.annotation.Autowired;
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.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
@Configuration
public class MqttInboundConfig {
    @Autowired
    private MqttConfig mqttConfig ;
    @Autowired
    private ReceiveMsgHandler receiveMsgHandler;
    /**
     * 配置消息接收通道
     * @return
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
    /**
     * 配置接收適配器
     */
    @Bean
    public MessageProducer messageProducer(MqttPahoClientFactory mqttPahoClientFactory) {
        MqttPahoMessageDrivenChannelAdapter adapter  =
                new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getUrl() ,
                        mqttConfig.getSubClientId() ,
                        mqttPahoClientFactory , mqttConfig.getSubTopic().split(",")) ;
        adapter.setConverter(new DefaultPahoMessageConverter());
        // 質(zhì)量服務等級
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter ;
    }
    /**
     * 配置接收消息處理器
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel") // 指定處理消息使用得通道
    public MessageHandler messageHandler() {
        return this.receiveMsgHandler ;
    }
}

發(fā)送消息配置類

MqttOutboundConfig.java

package com.study.mqtt.demo.outbound;
import com.study.mqtt.demo.domain.MqttConfig;
import org.springframework.beans.factory.annotation.Autowired;
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.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
@Configuration
public class MqttOutboundConfig {
    @Autowired
    private MqttConfig mqttConfig;
    @Autowired
    private MqttPahoClientFactory pahoClientFactory ;
    @Bean
    public MessageChannel mqttOutputChannel() {
        return new DirectChannel();
    }
    @Bean
    @ServiceActivator(inputChannel = "mqttOutputChannel")
    public MessageHandler mqttOutboundMassageHandler() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfig.getUrl() ,
                mqttConfig.getPubClientId() , pahoClientFactory ) ;
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(0);
        messageHandler.setDefaultTopic("default");
        return messageHandler ;
    }
}

發(fā)送消息網(wǎng)關接口類

MqttGateway.java

package com.study.mqtt.demo.gateway;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
@MessagingGateway(defaultRequestChannel = "mqttOutputChannel")
public interface MqttGateway {
    /**
     * 發(fā)送mqtt消息
     * @param topic 主題
     * @param payload 內(nèi)容
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
    /**
     * 發(fā)送包含qos的消息
     * @param topic 主題
     * @param qos 對消息處理的幾種機制。
     *          * 0 發(fā)送成功就算完成，會出現(xiàn)消息丟失
     *          * 1 增加消息重試機制，消息發(fā)送失敗會重新發(fā)送，會出現(xiàn)重復消息
     *          * 2 多了一次去重的動作，確保只有一次消息推給訂閱者。
     * @param payload 消息體
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
}

發(fā)送消息服務類

MqttMsgSenderService.java

package com.study.mqtt.demo.service;
import com.study.mqtt.demo.gateway.MqttGateway;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class MqttMsgSenderService {
    @Autowired
    private MqttGateway mqttGateway;
    public void send(String topic, String payload) {
        mqttGateway.sendToMqtt(topic, payload);
    }
    public void send(String topic, int qos, String payload) {
        mqttGateway.sendToMqtt(topic, qos, payload);
    }
}

測試驗證

訂閱消息驗證

啟動項目

發(fā)送消息

主題為配置文件中配置的訂閱主題 lq/iot/demo/
發(fā)送時間： 2025-05-25 21:29:26:439

訂閱收到消息

接收到消息的時間：Sun May 25 21:29:26 GMT+08:00 2025
???????接收到的主題：lq/iot/demo/
???????接收到的內(nèi)容：{ "msg":"spring boot mqtt demo" }

發(fā)送消息驗證

編寫測試類
- 發(fā)送主題：sb/mqtt/test
- ???????發(fā)送內(nèi)容：hello world ！=> 當前時間

package com.study.mqtt.demo;
import com.study.mqtt.demo.service.MqttMsgSenderService;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import java.util.Date;
@SpringBootTest(classes = MqttDemoApplication.class)
class MqttDemoApplicationTests {
    @Autowired
    private MqttMsgSenderService mqttMsgSenderService;
    @Test
    void contextLoads() {
    }
    @Test
    void sendMsg(){
        mqttMsgSenderService.send("sb/mqtt/test", "hello world ! => " + new Date());
    }
}

創(chuàng)建訂閱者

訂閱主題： sb/mqtt/test

運行測試類

訂閱者接收消息

主題：sb/mqtt/test

到此這篇關于MQTT SpringBoot整合的文章就介紹到這了,更多相關MQTT SpringBoot整合內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

springboot配置請求超時時間(Http會話和接口訪問)
本文主要介紹了springboot配置請求超時時間,包含Http會話和接口訪問兩種,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
2024-07-07
詳解SpringBoot如何實現(xiàn)多環(huán)境配置
在實際的軟件開發(fā)過程中，一個應用程序通常會有多個環(huán)境，pring?Boot?提供了一個非常靈活和強大的方式來管理這些環(huán)境配置，下面就跟隨小編一起學習一下吧
2023-07-07
一文學會如何在SpringBoot中使用線程池執(zhí)行定時任務
在開發(fā)現(xiàn)代應用程序時，定時任務是一項常見的需求，SpringBoot提供了一個強大的定時任務框架，可以輕松地執(zhí)行各種定時任務，結合線程池的使用，可以更好地管理任務的執(zhí)行，提高系統(tǒng)的性能和穩(wěn)定性，本文將介紹如何在Spring Boot中使用線程池執(zhí)行定時任務
2023-06-06
基于JavaMail的Java實現(xiàn)簡單郵件發(fā)送功能
這篇文章主要為大家詳細介紹了基于JavaMail的Java實現(xiàn)簡單郵件發(fā)送功能，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2017-09-09
Java設計模式之代理模式解析
這篇文章主要介紹了Java設計模式之代理模式解析,代理模式為一個對象提供一個替身,以控制對這個對象的訪問,即通過代理對象訪問目標對象,需要的朋友可以參考下
2023-12-12
Java檢查StringBuilder是否包含字符串的幾種方法
在Java中,檢查StringBuilder是否包含字符串（即非空）的最直接方法是檢查其長度,本文整理了一些判斷方法,感興趣的小伙伴可以了解一下
2025-06-06
Mybatis批量插入大量數(shù)據(jù)的最優(yōu)方式總結
批量插入功能是我們?nèi)粘９ぷ髦斜容^常見的業(yè)務功能之一,下面這篇文章主要給大家總結介紹了關于Mybatis批量插入大量數(shù)據(jù)的幾種最優(yōu)方式,文中通過實例代碼介紹的非常詳細,需要的朋友可以參考下
2023-03-03
MyBatis 批量新增工具類實現(xiàn)示例
本文主要介紹了MyBatis批量操作工具類MyBatisBatchExecutorUtil,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
2025-06-06
@PathVariable獲取路徑中帶有 / 斜杠的解決方案
這篇文章主要介紹了@PathVariable獲取路徑中帶有 / 斜杠的解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
2021-10-10
Java Mail郵件發(fā)送如何實現(xiàn)簡單封裝
這篇文章主要介紹了Java Mail郵件發(fā)送如何實現(xiàn)簡單封裝,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下
2020-11-11