mqtt调整：实现订阅多个mqtt服务

zouyu

昨天 69310a7de3d963c2bf46250b0965a2c7e8532f1e

 cnas-device/src/main/java/com/ruoyi/device/mqtt/MQCallback.java

@@ -1,15 +1,8 @@
package com.ruoyi.device.mqtt;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.ruoyi.device.constant.DCResistanceMqttConstants;
import com.ruoyi.device.service.CollectBridgeService;
import com.ruoyi.device.vo.DCResistanceMqttVO;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
@@ -17,17 +10,14 @@

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Objects;

@Component
@Slf4j
public class MQCallback<component> implements MqttCallback {
public class MQCallback implements MqttCallback {

    private MQClient mqClient; // MQTT连接数据
    private final MQClient mqClient; // MQTT连接数据

    private MQConfig mqConfig; // yml配置数据
    private final DeviceMqttConfig mqConfig; // yml配置数据

    private static MQCallback mqCallback;

@@ -41,7 +31,7 @@
      mqCallback.collectBridgeService = this.collectBridgeService;
    }

    public MQCallback(MQClient mqClient, MQConfig mqConfig) {
    public MQCallback(MQClient mqClient, DeviceMqttConfig mqConfig) {
        this.mqClient = mqClient;
        this.mqConfig = mqConfig;
    }
@@ -74,17 +64,24 @@
     * MQTT服务器向WEB服务器发送的数据会执行到这里面，官方话称为：订阅后的消息
     * @param topic 主题：也称为底层网关唯一标识
     * @param message 信息
     * @throws Exception 报错
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
    public void messageArrived(String topic, MqttMessage message) {
        try {
            String parse = new String(message.getPayload());
//            JSONObject jsonObject = JSONObject.parseObject(parse);
            switch (topic){
                case "/ztt/v3/2455220/publish":
                    //耐丝：直流电阻数据解析
                    mqCallback.collectBridgeService.addBridgeValueByNS(parse);
                    break;
                case "/ztt/v3/2455221/publish":
                    //耐丝：伸长率数据解析
                    log.info("伸长率消息体：{}",parse);
                    break;
            }
            // 填充采集数据
//            JSONObject jsonObject = JSONObject.parseObject(parse);
//            mqCallback.collectBridgeService.addBridgeValue(jsonObject);
            //耐丝：直流电阻数据解析
            mqCallback.collectBridgeService.addBridgeValueByNS(parse);

        } catch (Exception e) {
            e.printStackTrace();