springboot整合mqtt的步驟示例詳解

更新時間：2025年08月14日 14:46:49 作者：泓山

MQTT(Message Queuing Telemetry Transport)是一種輕量級的消息傳輸協議,適用于物聯網設備之間的通信,本文介紹Spring Boot整合MQTT的實現,涵蓋依賴引入、YML配置、配置類創(chuàng)建、自定義注解及使用示例,感興趣的朋友跟隨小編一起看看吧

1、引入依賴包

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-integration</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-stream</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>

2、yml配置

若需要搭建mqtt服務教程，留言我下期出哦！

spring:
  application:
    name: device-control
  profiles:
    active: local
device:
  mqtt:
    enable: true
    username: admin
    password: 123456
    host-url: tcp://192.168.1.12:1883         # mqtt服務連接tcp地址
    in-client-id: ${random.value}              # 隨機值，使出入站 client ID 不同
    out-client-id: ${random.value}
    client-id: ${random.int}                   # 客戶端Id，不能相同，采用隨機數 ${random.value}
    default-topic: pubDevice                      # 默認主題
    timeout: 60                                # 超時時間
    keepalive: 60                              # 保持連接
    clearSession: true                         # 清除會話(設置為false,斷開連接，重連后使用原來的會話 保留訂閱的主題，能接收離線期間的消息)

3、創(chuàng)建配置

創(chuàng)建MqttAutoConfiguration

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
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.scheduling.concurrent.ThreadPoolTaskExecutor;
import javax.annotation.Resource;
import java.util.concurrent.ThreadPoolExecutor;
@AutoConfiguration
@ConditionalOnProperty(value = "device.mqtt.enable", havingValue = "true")
@IntegrationComponentScan
public class MqttAutoConfiguration {
    @Resource
    MqttProperties mqttProperties;
    @Resource
    MqttMessageHandle mqttMessageHandle;
    /**
     * Mqtt 客戶端工廠 所有客戶端從這里產生
     * @return
     */
    @Bean
    public MqttPahoClientFactory mqttPahoClientFactory(){
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(mqttProperties.getHostUrl().split(","));
        options.setUserName(mqttProperties.getUsername());
        options.setPassword(mqttProperties.getPassword().toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }
    /**
     * Mqtt 管道適配器
     * @param factory
     * @return
     */
    @Bean
    public MqttPahoMessageDrivenChannelAdapter adapter(MqttPahoClientFactory factory){
        return new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getInClientId(),factory,mqttProperties.getDefaultTopic().split(","));
    }
    /**
     * 消息生產者 (接收,處理來自mqtt的消息)
     * @param adapter
     * @return
     */
    @Bean
    public IntegrationFlow mqttInbound(MqttPahoMessageDrivenChannelAdapter adapter) {
        adapter.setCompletionTimeout(5000);
        adapter.setQos(1);
        return IntegrationFlows.from( adapter)
                .channel(new ExecutorChannel(mqttThreadPoolTaskExecutor()))
                .handle(mqttMessageHandle)
                .get();
    }
    @Bean
    public ThreadPoolTaskExecutor mqttThreadPoolTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 最大可創(chuàng)建的線程數
        int maxPoolSize = 200;
        executor.setMaxPoolSize(maxPoolSize);
        // 核心線程池大小
        int corePoolSize = 50;
        executor.setCorePoolSize(corePoolSize);
        // 隊列最大長度
        int queueCapacity = 1000;
        executor.setQueueCapacity(queueCapacity);
        // 線程池維護線程所允許的空閑時間
        int keepAliveSeconds = 300;
        executor.setKeepAliveSeconds(keepAliveSeconds);
        // 線程池對拒絕任務(無線程可用)的處理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
    /**
     * 出站處理器 (向 mqtt 發(fā)送消息)
     * @param factory
     * @return
     */
    @Bean
    public IntegrationFlow mqttOutboundFlow(MqttPahoClientFactory factory) {
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(mqttProperties.getOutClientId(),factory);
        handler.setAsync(true);
        handler.setConverter(new DefaultPahoMessageConverter());
        handler.setDefaultTopic(mqttProperties.getDefaultTopic().split(",")[0]);
        return IntegrationFlows.from( "mqttOutboundChannel").handle(handler).get();
    }
}

創(chuàng)建MqttGateway

import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Lazy;
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;
@Component
@Lazy
@ConditionalOnProperty(value = "device.mqtt.enable", havingValue = "true")
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {
    /**
     * @param topic String
     * @param data  String
     * @return void
     * @throws
     * @description <description you method purpose>
     * @author lwt
     * @time 2024/1/24 09:29
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String data);
    /**
     * @param topic String
     * @param Qos   Integer
     * @param data  String
     * @return void
     * @throws
     * @description <description you method purpose>
     * @author lwt
     * @time 2024/1/24 09:31
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) Integer Qos, String data);
}

創(chuàng)建MqttMessageHandle

import cn.hutool.extra.spring.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Map;
@Slf4j
@AutoConfiguration
public class MqttMessageHandle implements MessageHandler {
    public static Map<String, Object> mqttServices;
    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        getMqttTopicService(message);
    }
    public Map<String, Object> getMqttServices() {
        if (mqttServices == null) {
            mqttServices = SpringUtil.getConfigurableBeanFactory().getBeansWithAnnotation(MqttService.class);
        }
        return mqttServices;
    }
    public void getMqttTopicService(Message<?> message) {
        // 在這里 我們根據不同的 主題 分發(fā)不同的消息
        String receivedTopic = message.getHeaders().get("mqtt_receivedTopic", String.class);
        if (receivedTopic == null || "".equals(receivedTopic)) {
            return;
        }
        //updateTopicStatus(receivedTopic);
        for (Map.Entry<String, Object> entry : getMqttServices().entrySet()) {
            // 把所有帶有 @MqttService 的類遍歷
            Class<?> clazz = entry.getValue().getClass();
            // 獲取他所有方法
            Method[] methods = clazz.getSuperclass().getDeclaredMethods();
            for (Method method : methods) {
                if (method.isAnnotationPresent(MqttTopic.class)) {
                    // 如果這個方法有 這個注解
                    MqttTopic handleTopic = method.getAnnotation(MqttTopic.class);
                    if (isMatch(receivedTopic, handleTopic.value())) {
                        // 并且 這個 topic 匹配成功
                        try {
                            method.invoke(SpringUtil.getBean(clazz),receivedTopic, message);
                            return;
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                            log.error("代理炸了");
                        } catch (InvocationTargetException e) {
                            log.error("執(zhí)行 {} 方法出現錯誤", handleTopic.value(), e);
                        }
                    }
                }
            }
        }
    }
    /**
     * mqtt 訂閱的主題與我實際的主題是否匹配
     * @param topic   是實際的主題
     * @param pattern 是我訂閱的主題 可以是通配符模式
     * @return 是否匹配
     */
    public static boolean isMatch(String topic, String pattern) {
        if ((topic == null) || (pattern == null)) {
            return false;
        }
        if (topic.equals(pattern)) {
            // 完全相等是肯定匹配的
            return true;
        }
        if ("#".equals(pattern)) {
            // # 號代表所有主題  肯定匹配的
            return true;
        }
        String[] splitTopic = topic.split("_");
        String[] splitPattern = pattern.split("_");
        boolean match = true;
        // 如果包含 # 則只需要判斷 # 前面的
        for (int i = 0; i < splitPattern.length; i++) {
            if (!"#".equals(splitPattern[i])) {
                // 不是# 號 正常判斷
                if (i >= splitTopic.length) {
                    // 此時長度不相等 不匹配
                    match = false;
                    break;
                }
                if (!splitTopic[i].equals(splitPattern[i]) && !"+".equals(splitPattern[i])) {
                    // 不相等 且不等于 +
                    match = false;
                    break;
                }
            } else {
                // 是# 號  肯定匹配的
                break;
            }
        }
        return match;
    }
}

創(chuàng)建MqttProperties

import lombok.Data;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.context.properties.ConfigurationProperties;
@ConfigurationProperties(prefix = "device.mqtt")
@Data
@AutoConfiguration
public class MqttProperties {
    /**
     * 用戶名
     */
    private String username;
    /**
     * 密碼
     */
    private String password;
    /**
     * 連接地址
     */
    private String hostUrl;
    /**
     * 進-客戶Id
     */
    private String inClientId;
    /**
     * 出-客戶Id
     */
    private String outClientId;
    /**
     * 客戶Id
     */
    private String clientId;
    /**
     * 默認連接話題
     */
    private String defaultTopic;
    /**
     * 超時時間
     */
    private int timeout;
    /**
     * 保持連接數
     */
    private int keepalive;
    /**是否清除session*/
    private boolean clearSession;
}

創(chuàng)建MqttConstants

public class MqttConstants {
    public static final String MQTT_DEVICE_INFO = "mqtt:device:info";
    public static final String TOPIC_PUB_DEVICE = "pubDevice";
    public static final String TOPIC_SUB_DEVICE = "subDevice";
}

創(chuàng)建初始化

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
@Slf4j
@Component
@ConditionalOnProperty(value = "device.mqtt.enable", havingValue = "true")
public class InitMqttSubscriberTopic {
    @Resource
    MqttSubscriberService mqttSubscriberService;
    @PostConstruct
    public void initSubscriber() {
        try {
            mqttSubscriberService.addTopic(MqttConstants.TOPIC_PUB_DEVICE);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
}

4、自定義注解

創(chuàng)建MqttService

import org.springframework.core.annotation.AliasFor;
import org.springframework.stereotype.Component;
import java.lang.annotation.*;
@Documented
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Component
public @interface MqttService {
    @AliasFor(annotation = Component.class)
    String value() default "";
}

創(chuàng)建MqttTopic

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface MqttTopic {
    /**
     * 主題名字
     */
    String value() default "";
}

創(chuàng)建如圖：

6、使用示例

import cn.hutool.extra.spring.SpringUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.Message;
import org.springframework.transaction.annotation.Transactional;
@Slf4j
@MqttService
public class MqttTopicHandle {
    /**
     * 監(jiān)聽到指定主題的消息
     * @param topic
     * @param message
     */
    @SneakyThrows
    @MqttTopic("pubDevice")
    @Transactional(rollbackFor = Exception.class)
    public void receive(String topic, Message<?> message) {
        log.info("message:{}", message.getPayload());
        String value = message.getPayload().toString();
        // 進行邏輯處理
    }
    /**
     * 發(fā)送消息到指定主題
     * @param topic
     * @param message
     */
    @Transactional(rollbackFor = Exception.class)
    public Boolean send(String topic, String message) {
        try {
            MqttGateway mqttGateway = SpringUtil.getBean(MqttGateway.class);
            mqttGateway.sendToMqtt(topic,message);
        } catch (Exception e){
            return false;
        }
        return true;
    }
}

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

您可能感興趣的文章:

解決線程異常WAITING(parking)問題
文章總結：在項目中線程數量持續(xù)增長無法回收,導致服務器卡死和內存不足,通過分析發(fā)現是由于自定義線程池未執(zhí)行shutdown()關閉導致線程泄漏
2024-11-11
SpringBoot實現插件化架構的4種方案詳解
在復雜業(yè)務場景下,傳統的單體應用架構往往面臨著功能擴展困難等困難,插件化架構作為一種模塊化設計思想的延伸,能夠使系統具備更好的擴展性和靈活性,下面我們來看看SpringBoot環(huán)境下實現插件化架構的4種實現方案吧
2025-05-05
java中Vector的詳細說明
Vector是Java早期的線程安全動態(tài)數組,支持自動擴容和有序可重復元素,因同步機制性能較低,已被ArrayList等替代,現代開發(fā)中,推薦使用ArrayList或并發(fā)容器而非Vector
2025-10-10
談談 Java 中 this 的使用方法
這篇文章主要介紹了Java 中 this 的使用方法，需要的朋友可以參考下
2014-01-01
TransmittableThreadLocal解決線程間上下文傳遞煩惱
這篇文章主要為大家介紹了TransmittableThreadLocal解決線程間上下文傳遞煩惱詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪
2022-11-11
菜鳥學習java設計模式之單例模式
這篇文章主要為大家詳細介紹了java設計模式之單例模式的相關資料，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2017-11-11
Java用itextpdf導出PDF方法(通俗易懂)
因為項目需要導出PDF文件,所以去找了一下能夠生成PDF的java工具,這篇文章主要給大家介紹了關于Java用itextpdf導出PDF的相關資料,文中介紹的方法通俗易懂,需要的朋友可以參考下
2023-07-07
javax.net.ssl.SSLHandshakeException:異常原因及解決方案
javax.net.ssl.SSLHandshakeException是一個SSL握手異常,通常在建立SSL連接時發(fā)生,這篇文章主要介紹了javax.net.ssl.SSLHandshakeException:異常原因及解決方案,需要的朋友可以參考下
2025-06-06
使用@CacheEvict清除指定下所有緩存
這篇文章主要介紹了使用@CacheEvict清除指定下所有緩存，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
2021-12-12
Java8新特性之類型注解_動力節(jié)點Java學院整理
這篇文章主要介紹了Java8新特性之類型注解的相關資料,需要的朋友可以參考下
2017-06-06