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

package com.ruoyi;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ruoyi.system.domain.mqtt.MqttBean;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.quartz.impl.StdSchedulerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @author 1097022316
 * 啟動后建立MQTT連接 并對數(shù)據(jù)持久化和相關(guān)邏輯處理
 */
@Component
public class StartInit implements ApplicationRunner {

    @Override
    public void run(ApplicationArguments args) throws Exception {
        MemoryPersistence persistence = new MemoryPersistence();
        
        //這里是從數(shù)據(jù)庫中查詢出所有的mqtt連接相關(guān)信息，如ip、topic等
        List<DeviceCollector> collectorList = collectorService.list(new QueryWrapper<DeviceCollector>().isNotNull("collector_ip").ne("collector_ip", "").eq("is_del", "0").groupBy("collector_topic"));
        
        System.err.println("collectorList = " + collectorList);
        
        //如果沒有任何連接 直接結(jié)束
        if (CollUtil.isEmpty(collectorList)) {
            return;
        }
        
        List<MqttBean> mqttBeans = new ArrayList<>();
        
        //把從數(shù)據(jù)庫中查詢出來的信息組裝成mqttclient連接所需要的對象
        //一般都是ip、port、username、password、topic、clientid 這里是簡單的用法，如有高級用法可自行摸索（順便在下面評論教一下）
        
        collectorList.forEach(mqtt -> {
            MqttBean bean = new MqttBean("tcp://" + mqtt.getCollectorIp() + ":" + mqtt.getCollectorPort(), mqtt.getCollectorUsername(), mqtt.getCollectorPassword(), mqtt.getCollectorTopic(), mqtt.getCollectorClientId());
            mqttBeans.add(bean);
        });

        //對我們組裝的mqtt連接對象信息進(jìn)行遍歷循環(huán)連接
        mqttBeans.forEach(bean -> {
            try {
                    MqttClient mqttClient = new MqttClient(bean.getUrl(), bean.getClientId(), persistence);
                    MqttConnectOptions connOpts = new MqttConnectOptions();
                	//設(shè)置相關(guān)連接參數(shù)，有些是必要有些是非必要 可自行點進(jìn)去查看源碼
                    connOpts.setAutomaticReconnect(false);
                    connOpts.setCleanSession(true);
                    connOpts.setUserName(bean.getUserName());
                    connOpts.setPassword(bean.getPassword().toCharArray());
                    mqttClient.connect(connOpts);
                    //把連接對象加入到全局
                    mqttClients.put(bean.getTopics(), mqttClient);

                    mqttClient.setCallback(new MqttCallbackExtended() {

                        @Override
                        public void connectComplete(boolean reconnect, String serverURI) {
                            if (reconnect) {
                                System.out.println("Reconnected successfully.  url = " + serverURI);
                            } else {
                                System.out.println("Connected successfully for the first time.");
                            }
                        }

                        /**
                         * 設(shè)置重連機(jī)制
                         */
                        @Override
                        public void connectionLost(Throwable cause) {
                            System.err.println(bean.getTopics() + "連接丟失" + cause.getMessage());
                            if (!mqttClient.isConnected()) {
                                try {
                                    Thread.sleep(1000 * 60 * 5);
                                    //嘗試連接
                                    System.out.println("bean = " + bean);
                                    boolean flag = MyMqttUtils.connectMqtt(new MqttBean(bean.getUrl(), bean.getUserName(), bean.getPassword(), bean.getTopics(), bean.getClientId()));
                                    if (flag) {
                                        System.err.println(bean.getTopics() + "重新連接，重新訂閱！");
                                    }
                                } catch (InterruptedException e) {
                                    MyMqttUtils.saveToTxt(bean.getTopics() + "MQTT連接出異常了" + e.getMessage(), "CLCW");
                                    throw new RuntimeException(e);
                                }
                            }
                        }

                        @Override
                        public void messageArrived(String topic, MqttMessage message) {
							//這里是當(dāng)消息推送時我們做的事情
                            System.out.println("dosomethings...");
                        }

                        @Override
                        public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                            System.out.println("消息發(fā)送完整");
                        }
                    });
                    mqttClient.subscribe(bean.getTopics(), 2);

            } catch (Exception e) {
                e.printStackTrace();
                System.err.println(bean.getTopics() + "MQTT連接出異常了");
                MyMqttUtils.saveToTxt(bean.getTopics() + "MQTT連接出異常了"+e.getMessage(),"LJCW");
                try {
                    Thread.sleep(1000 * 60 * 30);
                    //嘗試連接
                    System.out.println("bean = " + bean);
                    boolean flag = MyMqttUtils.connectMqtt(new MqttBean(bean.getUrl(), bean.getUserName(), bean.getPassword(), bean.getTopics(), bean.getClientId()));
                    if (flag) {
                        System.err.println(bean.getTopics() + "重新連接，重新訂閱！");
                    }
                } catch (InterruptedException ex) {
                    //可以在這里把報錯信息存入本地看看
                    throw new RuntimeException(ex);
                }
            }
        });
    }
}

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author 1097022316
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public class MqttBean {

    private String url;

    private String userName;

    private String password;

    private String topics;

    private String clientId;

}