springboot整合mqtt的步驟示例詳解

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

MQTT(Message Queuing Telemetry Transport)是一種輕量級(jí)的消息傳輸協(xié)議,適用于物聯(lián)網(wǎng)設(shè)備之間的通信,本文介紹Spring Boot整合MQTT的實(shí)現(xiàn),涵蓋依賴引入、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服務(wù)教程，留言我下期出哦！

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服務(wù)連接tcp地址
    in-client-id: ${random.value}              # 隨機(jī)值，使出入站 client ID 不同
    out-client-id: ${random.value}
    client-id: ${random.int}                   # 客戶端Id，不能相同，采用隨機(jī)數(shù) ${random.value}
    default-topic: pubDevice                      # 默認(rèn)主題
    timeout: 60                                # 超時(shí)時(shí)間
    keepalive: 60                              # 保持連接
    clearSession: true                         # 清除會(huì)話(設(shè)置為false,斷開(kāi)連接，重連后使用原來(lái)的會(huì)話 保留訂閱的主題，能接收離線期間的消息)

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 客戶端工廠 所有客戶端從這里產(chǎn)生
     * @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(","));
    }
    /**
     * 消息生產(chǎn)者 (接收,處理來(lái)自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)建的線程數(shù)
        int maxPoolSize = 200;
        executor.setMaxPoolSize(maxPoolSize);
        // 核心線程池大小
        int corePoolSize = 50;
        executor.setCorePoolSize(corePoolSize);
        // 隊(duì)列最大長(zhǎng)度
        int queueCapacity = 1000;
        executor.setQueueCapacity(queueCapacity);
        // 線程池維護(hù)線程所允許的空閑時(shí)間
        int keepAliveSeconds = 300;
        executor.setKeepAliveSeconds(keepAliveSeconds);
        // 線程池對(duì)拒絕任務(wù)(無(wú)線程可用)的處理策略
        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) {
        // 在這里 我們根據(jù)不同的 主題 分發(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)) {
                    // 如果這個(gè)方法有 這個(gè)注解
                    MqttTopic handleTopic = method.getAnnotation(MqttTopic.class);
                    if (isMatch(receivedTopic, handleTopic.value())) {
                        // 并且 這個(gè) topic 匹配成功
                        try {
                            method.invoke(SpringUtil.getBean(clazz),receivedTopic, message);
                            return;
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                            log.error("代理炸了");
                        } catch (InvocationTargetException e) {
                            log.error("執(zhí)行 {} 方法出現(xiàn)錯(cuò)誤", handleTopic.value(), e);
                        }
                    }
                }
            }
        }
    }
    /**
     * mqtt 訂閱的主題與我實(shí)際的主題是否匹配
     * @param topic   是實(shí)際的主題
     * @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)) {
            // # 號(hào)代表所有主題  肯定匹配的
            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])) {
                // 不是# 號(hào) 正常判斷
                if (i >= splitTopic.length) {
                    // 此時(shí)長(zhǎng)度不相等 不匹配
                    match = false;
                    break;
                }
                if (!splitTopic[i].equals(splitPattern[i]) && !"+".equals(splitPattern[i])) {
                    // 不相等 且不等于 +
                    match = false;
                    break;
                }
            } else {
                // 是# 號(hào)  肯定匹配的
                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;
    /**
     * 進(jìn)-客戶Id
     */
    private String inClientId;
    /**
     * 出-客戶Id
     */
    private String outClientId;
    /**
     * 客戶Id
     */
    private String clientId;
    /**
     * 默認(rèn)連接話題
     */
    private String defaultTopic;
    /**
     * 超時(shí)時(shí)間
     */
    private int timeout;
    /**
     * 保持連接數(shù)
     */
    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)聽(tīng)到指定主題的消息
     * @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();
        // 進(jìn)行邏輯處理
    }
    /**
     * 發(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;
    }
}

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

您可能感興趣的文章: