springboot集成mqtt超級詳細步驟

更新時間：2023年06月28日 16:42:11 作者：想要一百塊

這篇文章主要介紹了springboot集成mqtt超級詳細步驟，本文分步驟結(jié)合實例代碼給大家介紹的非常詳細，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下

springboot集成MQTT步驟

1. 引入pom依賴

   <!-- mqtt -->
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-mqtt</artifactId>
    </dependency>

2. application.yml

## MQTT##
mqtt:
    host: tcp://192.168.10.198:1883
    userName: root
    passWord: 123456
    qos: 1
    clientId: ClientId_local #ClientId_local必須唯一 比如你已經(jīng)定了叫ABC  那你就一直叫ABC  其他地方就不要使用ABC了
    timeout: 10
    keepalive: 20
    topic1: A/pick/warn/#  #符號是代表整個warn下面的全部子主題 沒有理解的話 可以百度仔細理解一下
    topic2: A/cmd/resp
    topic3: ABCF
    topic4: ABCH

application.properties

## MQTT##
mqtt.host=tcp://192.168.10.198:1883
mqtt.clientId=ClientId_local
mqtt.username=admin
mqtt.password=123456
mqtt.timeout=10
mqtt.keepalive=20
mqtt.topic1=A/pick/warn/#

3. MqttConfiguration.java

import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 1. @author WXY
 2. @date 2022/6/29 20:42
 */
@Configuration
public class MqttConfiguration {
    private static final Logger log = LoggerFactory.getLogger(MqttConfiguration.class);
    @Value("${mqtt.host}")
    String host;
    @Value("${mqtt.username}")
    String username;
    @Value("${mqtt.password}")
    String password;
    @Value("${mqtt.clientId}")
    String clientId;
    @Value("${mqtt.timeout}")
    int timeOut;
    @Value("${mqtt.keepalive}")
    int keepAlive;
    @Value("${mqtt.topic1}")
    String topic1;
    @Value("${mqtt.topic2}")
    String topic2;
    @Value("${mqtt.topic3}")
    String topic3;
    @Value("${mqtt.topic4}")
    String topic4;
    @Bean//注入spring
    public MyMQTTClient myMQTTClient() {
        MyMQTTClient myMQTTClient = new MyMQTTClient(host, username, password, clientId, timeOut, keepAlive);
        for (int i = 0; i < 10; i++) {
            try {
                myMQTTClient.connect();
                //不同的主題
             //   myMQTTClient.subscribe(topic1, 1);
             //   myMQTTClient.subscribe(topic2, 1);
             //   myMQTTClient.subscribe(topic3, 1);
             //   myMQTTClient.subscribe(topic4, 1);
                return myMQTTClient;
            } catch (MqttException e) {
                log.error("MQTT connect exception,connect time = " + i);
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        }
        return myMQTTClient;
    }
     public String getTopic1() {
        return topic1;
    }
    public void setTopic1(String topic1) {
        this.topic1 = topic1;
    }
    public String getTopic2() {
        return topic2;
    }
    public void setTopic2(String topic2) {
        this.topic2 = topic2;
    }
    public String getTopic3() {
        return topic3;
    }
    public void setTopic3(String topic3) {
        this.topic3 = topic3;
    }
    public String getTopic4() {
        return topic4;
    }
    public void setTopic4(String topic4) {
        this.topic4 = topic4;
    }
}

4. MyMQTTClient.java

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 1. @author WXY
 2. @date 2022/6/29 20:43
 */
public class MyMQTTClient {
    private static final Logger LOGGER = LoggerFactory.getLogger(MyMQTTClient.class);
    private static MqttClient client;
    private String host;
    private String username;
    private String password;
    private String clientId;
    private int timeout;
    private int keepalive;
    public MyMQTTClient(String host, String username, String password, String clientId, int timeOut, int keepAlive) {
        this.host = host;
        this.username = username;
        this.password = password;
        this.clientId = clientId;
        this.timeout = timeOut;
        this.keepalive = keepAlive;
    }
    public static MqttClient getClient() {
        return client;
    }
    public static void setClient(MqttClient client) {
        MyMQTTClient.client = client;
    }
    /**
     * 設(shè)置mqtt連接參數(shù)
     *
     * @param username
     * @param password
     * @param timeout
     * @param keepalive
     * @return
     */
    public MqttConnectOptions setMqttConnectOptions(String username, String password, int timeout, int keepalive) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        return options;
    }
    /**
     * 連接mqtt服務(wù)端，得到MqttClient連接對象
     */
    public void connect() throws MqttException {
        if (client == null) {
            client = new MqttClient(host, clientId, new MemoryPersistence());
            client.setCallback(new MyMQTTCallback(MyMQTTClient.this));
        }
        MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(username, password, timeout, keepalive);
        if (!client.isConnected()) {
            client.connect(mqttConnectOptions);
        } else {
            client.disconnect();
            client.connect(mqttConnectOptions);
        }
        LOGGER.info("MQTT connect success");//未發(fā)生異常，則連接成功
    }
    /**
     * 發(fā)布，默認qos為0，非持久化
     *
     * @param pushMessage
     * @param topic
     */
    public void publish(String pushMessage, String topic) {
        publish(pushMessage, topic, 0, false);
    }
    /**
     * 發(fā)布消息
     *
     * @param pushMessage
     * @param topic
     * @param qos
     * @param retained:留存
     */
    public void publish(String pushMessage, String topic, int qos, boolean retained) {
        MqttMessage message = new MqttMessage();
        message.setPayload(pushMessage.getBytes());
        message.setQos(qos);
        message.setRetained(retained);
        MqttTopic mqttTopic = MyMQTTClient.getClient().getTopic(topic);
        if (null == mqttTopic) {
            LOGGER.error("topic is not exist");
        }
        MqttDeliveryToken token;//Delivery:配送
        synchronized (this) {//注意：這里一定要同步，否則，在多線程publish的情況下，線程會發(fā)生死鎖，分析見文章最后補充
            try {
                token = mqttTopic.publish(message);//也是發(fā)送到執(zhí)行隊列中，等待執(zhí)行線程執(zhí)行，將消息發(fā)送到消息中間件
                token.waitForCompletion(1000L);
            } catch (MqttPersistenceException e) {
                e.printStackTrace();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }
    /**
     * 訂閱某個主題
     *
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        try {
            MyMQTTClient.getClient().subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    /**
     * 取消訂閱主題
     *
     * @param topic 主題名稱
     */
    public void cleanTopic(String topic) {
        if (client != null && client.isConnected()) {
            try {
                client.unsubscribe(topic);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        } else {
            System.out.println("取消訂閱失??！");
        }
    }
}

5. MyMQTTCallback.java

import cn.hutool.core.util.CharsetUtil;
import com.alibaba.fastjson.JSON;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
/**
 * @author WXY
 * @date 2022/6/29 20:43
 */
public class MyMQTTCallback implements MqttCallbackExtended {
    //手動注入
    private MqttConfiguration mqttConfiguration = SpringUtils.getBean(MqttConfiguration.class);
    private static final Logger log = LoggerFactory.getLogger(MyMQTTCallback.class);
    private MyMQTTClient myMQTTClient;
    public MyMQTTCallback(MyMQTTClient myMQTTClient) {
        this.myMQTTClient = myMQTTClient;
    }
   /**
     * 丟失連接，可在這里做重連
     * 只會調(diào)用一次
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.error("mqtt connectionLost 連接斷開，5S之后嘗試重連: {}", throwable.getMessage());
        long reconnectTimes = 1;
        while (true) {
            try {
                if (MyMQTTClient.getClient().isConnected()) {
                    //判斷已經(jīng)重新連接成功  需要重新訂閱主題 可以在這個if里面訂閱主題  或者 connectComplete（方法里面）  看你們自己選擇
                    log.warn("mqtt reconnect success end  重新連接  重新訂閱成功");
                    return;
                }
                reconnectTimes+=1;
                log.warn("mqtt reconnect times = {} try again...  mqtt重新連接時間 {}", reconnectTimes, reconnectTimes);
                MyMQTTClient.getClient().reconnect();
            } catch (MqttException e) {
                log.error("mqtt斷連異常", e);
            }
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e1) {
            }
        }
    }
    /**
     * @param topic
     * @param mqttMessage
     * @throws Exception
     * subscribe后得到的消息會執(zhí)行到這里面
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        log.info("接收消息主題 : {}，接收消息內(nèi)容 : {}", topic, new String(mqttMessage.getPayload()));
        //發(fā)布消息主題
        if (topic.equals("embed/resp")){
            Map maps = (Map) JSON.parse(new String(mqttMessage.getPayload(), CharsetUtil.UTF_8));
            //你自己的業(yè)務(wù)接口
            insertCmdResults(maps);
        }
        //接收報警主題
        if (topic.equals("embed/warn")){
            Map maps = (Map) JSON.parse(new String(mqttMessage.getPayload(), CharsetUtil.UTF_8));
            //你自己的業(yè)務(wù)接口
            insertPushAlarm(maps);
        }
    }
     /**
     *連接成功后的回調(diào) 可以在這個方法執(zhí)行 訂閱主題  生成Bean的 MqttConfiguration方法中訂閱主題 出現(xiàn)bug 
     *重新連接后  主題也需要再次訂閱  將重新訂閱主題放在連接成功后的回調(diào) 比較合理
     * @param reconnect
     * @param serverURI
     */
    @Override
    public  void  connectComplete(boolean reconnect,String serverURI){
        log.info("MQTT 連接成功，連接方式：{}",reconnect?"重連":"直連");
        //訂閱主題
        myMQTTClient.subscribe(mqttConfiguration.topic1, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic2, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic3, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic4, 1);
    }
    /**
     * 消息到達后
     * subscribe后，執(zhí)行的回調(diào)函數(shù)
     *
     * @param s
     * @param mqttMessage
     * @throws Exception
     */
    /**
     * publish后，配送完成后回調(diào)的方法
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());
    }
  }

6. MqttMsg.java

/**
 * @author WXY
 * @date 2022/6/29 20:44
 */
public class MqttMsg {
    private String name = "";
    private String content = "";
    private String time = "";
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String getContent() {
        return content;
    }
    public void setContent(String content) {
        this.content = content;
    }
    public String getTime() {
        return time;
    }
    public void setTime(String time) {
        this.time = time;
    }
    @Override
    public String toString() {
        return "MqttMsg{" +
                "name='" + name + '\'' +
                ", content='" + content + '\'' +
                ", time='" + time + '\'' +
                '}';
    }
}

7. MqttController.java

import com.gjwl.common.core.mqtt.MqttMsg;
import com.gjwl.common.core.mqtt.MyMQTTClient;
import net.sf.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.LinkedList;
import java.util.Queue;
/**
 * @author WXY
 * @date 2022/6/29 20:44
 */
@RestController
@RequestMapping("/sun/mqtt")
public class MqttController {
    @Autowired
    private MyMQTTClient myMQTTClient;
    @Value("${mqtt.topic1}")
    String topic1;
    @Value("${mqtt.topic2}")
    String topic2;
    @Value("${mqtt.topic3}")
    String topic3;
    @Value("${mqtt.topic4}")
    String topic4;
    Queue<String> msgQueue = new LinkedList<>();
    int count = 1;
    @PostMapping("/getMsg")
    @ResponseBody
    public synchronized void mqttMsg(MqttMsg mqttMsg) {
        System.out.println("隊列元素數(shù)量：" + msgQueue.size());
        System.out.println("***************" + mqttMsg.getName() + ":" + mqttMsg.getContent() + "****************");
        //時間格式化
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String time = df.format(new Date());
        mqttMsg.setTime(time);
        mqttMsg.setContent(mqttMsg.getContent() + "——后臺編號：" + count);
        count++;
        //map轉(zhuǎn)json
        JSONObject json = JSONObject.fromObject(mqttMsg);
        String sendMsg = json.toString();
        System.out.println(sendMsg);
        //隊列添加元素
        boolean flag = msgQueue.offer(sendMsg);
        if (flag) {
            //發(fā)布消息  topic2 是你要發(fā)送到那個通道里面的主題 比如我要發(fā)送到topic2主題消息
            myMQTTClient.publish(msgQueue.poll(), topic2);
            System.out.println("時間戳" + System.currentTimeMillis());
        }
        System.out.println("隊列元素數(shù)量：" + msgQueue.size());
    }
}

8.SpringUtils.java

import org.springframework.aop.framework.AopContext;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import com.gjwl.common.utils.StringUtils;
/**
 * spring工具類 方便在非spring管理環(huán)境中獲取bean
 * 
 * @author wxy
 */
@Component
public final class SpringUtils implements BeanFactoryPostProcessor, ApplicationContextAware 
{
    /** Spring應(yīng)用上下文環(huán)境 */
    private static ConfigurableListableBeanFactory beanFactory;
    private static ApplicationContext applicationContext;
    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException 
    {
        SpringUtils.beanFactory = beanFactory;
    }
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException 
    {
        SpringUtils.applicationContext = applicationContext;
    }
    /**
     * 獲取對象
     *
     * @param name
     * @return Object 一個以所給名字注冊的bean的實例
     * @throws org.springframework.beans.BeansException
     *
     */
    @SuppressWarnings("unchecked")
    public static <T> T getBean(String name) throws BeansException
    {
        return (T) beanFactory.getBean(name);
    }
    /**
     * 獲取類型為requiredType的對象
     *
     * @param clz
     * @return
     * @throws org.springframework.beans.BeansException
     *
     */
    public static <T> T getBean(Class<T> clz) throws BeansException
    {
        T result = (T) beanFactory.getBean(clz);
        return result;
    }
    /**
     * 如果BeanFactory包含一個與所給名稱匹配的bean定義，則返回true
     *
     * @param name
     * @return boolean
     */
    public static boolean containsBean(String name)
    {
        return beanFactory.containsBean(name);
    }
    /**
     * 判斷以給定名字注冊的bean定義是一個singleton還是一個prototype。 如果與給定名字相應(yīng)的bean定義沒有被找到，將會拋出一個異常（NoSuchBeanDefinitionException）
     *
     * @param name
     * @return boolean
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     *
     */
    public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException
    {
        return beanFactory.isSingleton(name);
    }
    /**
     * @param name
     * @return Class 注冊對象的類型
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     *
     */
    public static Class<?> getType(String name) throws NoSuchBeanDefinitionException
    {
        return beanFactory.getType(name);
    }
    /**
     * 如果給定的bean名字在bean定義中有別名，則返回這些別名
     *
     * @param name
     * @return
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     *
     */
    public static String[] getAliases(String name) throws NoSuchBeanDefinitionException
    {
        return beanFactory.getAliases(name);
    }
    /**
     * 獲取aop代理對象
     * 
     * @param invoker
     * @return
     */
    @SuppressWarnings("unchecked")
    public static <T> T getAopProxy(T invoker)
    {
        return (T) AopContext.currentProxy();
    }
    /**
     * 獲取當(dāng)前的環(huán)境配置，無配置返回null
     *
     * @return 當(dāng)前的環(huán)境配置
     */
    public static String[] getActiveProfiles()
    {
        return applicationContext.getEnvironment().getActiveProfiles();
    }
    /**
     * 獲取當(dāng)前的環(huán)境配置，當(dāng)有多個環(huán)境配置時，只獲取第一個
     *
     * @return 當(dāng)前的環(huán)境配置
     */
    public static String getActiveProfile()
    {
        final String[] activeProfiles = getActiveProfiles();
        return StringUtils.isNotEmpty(activeProfiles) ? activeProfiles[0] : null;
    }
}