添加mqtt故障上报开关

zhuguifei

2024-12-12 ca1d4a60b4493b0925552791da5effb42962fa39

添加mqtt故障上报开关

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/controller/DryRealTimeDataController.java

@@ -1,14 +1,9 @@
package org.jeecg.modules.dry.controller;


import ai.djl.modality.Classifications;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.druid.support.json.JSONUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
@@ -16,13 +11,9 @@
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.jeecg.common.api.vo.Result;

import org.jeecg.common.config.TenantContext;
import org.jeecg.common.constant.MqttConstant;
import org.jeecg.common.system.query.QueryGenerator;
import org.jeecg.common.util.IpUtils;
import org.jeecg.common.util.SpringContextUtils;
import org.jeecg.common.util.oConvertUtils;
import org.jeecg.config.mybatis.MybatisPlusSaasConfig;
import org.jeecg.modules.dry.entity.*;
import org.jeecg.modules.dry.mqtt.MqttConfig;
import org.jeecg.modules.dry.mqtt.MqttUtil;
@@ -33,17 +24,13 @@

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;

import javax.servlet.http.HttpServletRequest;
import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Marshaller;
import java.io.InputStream;
import java.io.StringWriter;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;


@Api(tags = "实时数据处理控制器")
@@ -100,7 +87,12 @@
                mqttMessage.setPayload(JSONObject.toJSONString(realTimeDataVo).getBytes());
                mqttUtil.getMqttClient().publish(MqttConstant.TENANT_UP_PREFIX_REALTIME_DATA,mqttMessage);
                //处理故障信息
                dryRealTimeDataService.fitFultRecord(realTimeDataVo);
                dryRealTimeDataService.fitFaultRecord(realTimeDataVo);
            }

            if ("user".equals(mqttConfig.getRole())){
                //处理故障信息
                dryRealTimeDataService.fitFaultRecord(realTimeDataVo);
            }

        } catch (MqttException e) {

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/service/IDryRealTimeDataService.java

@@ -21,5 +21,5 @@
     * @param realTimeDataVo
     * @return
     */
    Result<?> fitFultRecord(RealTimeDataVo realTimeDataVo);
    Result<?> fitFaultRecord(RealTimeDataVo realTimeDataVo);
}

 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/service/impl/DryRealTimeDataServiceImpl.java

@@ -71,6 +71,9 @@
    @Autowired
    private MqttUtil mqttUtil;

    @Value(value = "${jeecg.mqtt.enable}")
    private boolean mqttEnable;

    public String getTemporaryToken() {
        if (token == null) {
            RedisUtil redisUtil = SpringContextUtils.getBean(RedisUtil.class);
@@ -480,7 +483,7 @@
    }

    @Override
    public Result<?> fitFultRecord(RealTimeDataVo vo) {
    public Result<?> fitFaultRecord(RealTimeDataVo vo) {
        TenantContext.setTenant(vo.getTenantid()+"");
        ThreadUtil.execute(() -> {
            try {
@@ -492,7 +495,7 @@

                //处理结束后，将redis中实时数据发送至云服务器
                    Map<Object, Object> toCloudFaultMap = redisUtil.hmget(MqttConstant.MQTT_REAL_FAULT);
                    if(!toCloudFaultMap.isEmpty()){
                    if(mqttEnable && !toCloudFaultMap.isEmpty()){
                        MqMessage< Map<Object, Object>> message = new MqMessage<>();
                        message.setData(toCloudFaultMap);
                        message.setTentId(vo.getTenantid()+"");