干燥机配套车间生产管理系统/云平台服务端
blame | 历史 | 补丁 | 提交 | 提交对比 | 忽略空白
zhuguifei
2024-11-20 426f50baf57ae5901a6b5a26d1567e3b270d2499 
 jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/service/impl/DryRealTimeDataServiceImpl.java


@@ -3,48 +3,35 @@


import cn.hutool.core.bean.BeanUtil;


import cn.hutool.core.thread.ThreadUtil;


import com.alibaba.fastjson.JSON;


import com.alibaba.fastjson.JSONArray;


import com.alibaba.fastjson.JSONObject;


import com.alibaba.fastjson.TypeReference;


import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;


import lombok.extern.slf4j.Slf4j;


import org.apache.commons.lang3.StringUtils;


import org.apache.mina.core.service.IoAcceptor;


import org.apache.mina.core.service.IoHandler;


import org.apache.mina.core.session.IoSession;


import org.apache.shiro.SecurityUtils;


import org.eclipse.paho.client.mqttv3.MqttClient;


import org.eclipse.paho.client.mqttv3.MqttException;


import org.eclipse.paho.client.mqttv3.MqttMessage;


import org.jeecg.common.api.CommonAPI;


import org.jeecg.common.api.vo.Result;


import org.jeecg.common.config.TenantContext;


import org.jeecg.common.config.mqtoken.UserTokenContext;


import org.jeecg.common.constant.CommonCacheConstant;


import org.jeecg.common.constant.CommonConstant;


import org.jeecg.common.constant.MqttConstant;


import org.jeecg.common.system.util.JwtUtil;


import org.jeecg.common.system.vo.LoginUser;


import org.jeecg.common.util.DateUtils;


import org.jeecg.common.util.RedisUtil;


import org.jeecg.common.util.SpringContextUtils;


import org.jeecg.modules.dry.common.CacheConstants;


import org.jeecg.modules.dry.entity.*;


import org.jeecg.modules.dry.mqtt.MqMessage;


import org.jeecg.modules.dry.mqtt.MqttUtil;


import org.jeecg.modules.dry.service.*;


import org.jeecg.modules.dry.socket.ServerHandler;


import org.jeecg.modules.dry.socket.SocketServerConfig;


import org.jeecg.modules.dry.util.DryUtil;


import org.jeecg.modules.dry.vo.*;


import org.jetbrains.annotations.NotNull;


import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.beans.factory.annotation.Value;


import org.springframework.stereotype.Service;


import org.springframework.transaction.annotation.Transactional;




import javax.security.auth.login.LoginContext;


import java.io.IOException;


import java.io.ObjectOutputStream;


import java.net.Socket;


import java.text.DecimalFormat;


import java.util.*;


import java.util.stream.Collectors;


@@ -494,20 +481,41 @@




    @Override


    public Result<?> fitFultRecord(RealTimeDataVo vo) {


        TenantContext.setTenant(vo.getTenantid()+"");


        ThreadUtil.execute(() -> {


            try {


                //解析存储报警数据


                List<DryFaultRecord> faultRecords1 = fitFault(vo.getEqp_fault(), vo.getWorkorder(), vo.getTenantid(), vo.getMachineid(), 1);


                List<DryFaultRecord> faultRecords2 = fitFault(vo.getEqp_warning(), vo.getWorkorder(), vo.getTenantid(), vo.getMachineid(), 2);


                faultRecords1.addAll(faultRecords2);


                JSONObject json = new JSONObject();


                json.put("data",JSON.toJSONString(faultRecords1));




                if(faultRecords1.isEmpty())  return;


                MqttMessage mqttMessage = new MqttMessage();


                mqttMessage.setQos(0);


                mqttMessage.setPayload((JSON.toJSONString(json).getBytes()));


                mqttUtil.getMqttClient().publish(MqttConstant.TENANT_UP_PREFIX_FAULT_DATA,mqttMessage);




                //处理结束后，将redis中实时数据发送至云服务器


                    Map<Object, Object> toCloudFaultMap = redisUtil.hmget(MqttConstant.MQTT_REAL_FAULT);


                    if(!toCloudFaultMap.isEmpty()){


                        MqMessage< Map<Object, Object>> message = new MqMessage<>();


                        message.setData(toCloudFaultMap);


                        message.setTentId(vo.getTenantid()+"");


                        MqttMessage mqttMessage = new MqttMessage();


                        mqttMessage.setQos(0);


                        mqttMessage.setPayload(JSON.toJSONString(message).getBytes());


                        mqttUtil.getMqttClient().publish(MqttConstant.TENANT_UP_PREFIX_REAL_FAULT_DATA,mqttMessage);


                    }






                //要保存的历史故障


                if(!faultRecords1.isEmpty()){


                    MqMessage<List<DryFaultRecord>> message = new MqMessage<>();


                    message.setData(faultRecords1);


                    message.setTentId(vo.getTenantid()+"");


                    MqttMessage mqttMessage = new MqttMessage();


                    mqttMessage.setQos(0);


                    mqttMessage.setPayload((JSON.toJSONString(message).getBytes()));


                    mqttUtil.getMqttClient().publish(MqttConstant.TENANT_UP_PREFIX_FAULT_DATA,mqttMessage);


                }










            } catch (Exception e) {


                e.printStackTrace();


@@ -533,7 +541,7 @@


        //数据样本："eqp_fault": "滚筒降超时-报警,风机过流报警,滚筒升超时-报警,风箱升报警",


        System.err.println((faultType == 1 ? "类型：故障" : "类型：报警") +  DateUtils.formatDateTime()+"--"+fault);


        //redis中的故障


        Map<Object, Object> rFauMap = redisUtil.hmget(MqttConstant.MQTT_EQP_FAULT);


        Map<Object, Object> rFauMap = redisUtil.hmget(MqttConstant.MQTT_REAL_FAULT);


        Map<String, Object> redFauMap = rFauMap.entrySet().stream()


                .collect(Collectors.toMap(


                        entry -> entry.getKey().toString(),  // 键转换为字符串


@@ -550,50 +558,74 @@


        }


        //1.解析数据


        String[] eqpFaults = fault.split(",");


        Map<String,DryFaultRecord> addFauMap = new HashMap<>();


        Map<String,DryFaultRecord> realFauMap = new HashMap<>();


        for (int i = 0; i < eqpFaults.length; i++) {


            String eqpFault = eqpFaults[i];


            //避免空字符串


            if(StringUtils.isEmpty(eqpFault)) continue;


            if(StringUtils.isEmpty(eqpFault.trim())) continue;


            //1.1检查mqtt中是否已存在这个故障


            String redisKey = String.format("%s_%s_%s", tenantId, machineId,eqpFault);


            String  rFault = (String) redisUtil.get(redisKey);


            String redisKey = String.format("%s_%s_%s", tenantId, machineId,eqpFault).trim();






            realFauMap.put(redisKey, new DryFaultRecord());


            DryFaultRecordVo  rFault = (DryFaultRecordVo) redisUtil.hget(MqttConstant.MQTT_REAL_FAULT,redisKey);


            //1.2如果redis不存在则存入（存故障开始）


            if(rFault ==null){


                //组装缓存数据


                DryFaultRecord faultRecord = new DryFaultRecord(orderId,eqpFault,faultType,new Date(),null);


                realFauMap.put(redisKey,faultRecord);


//                DryFaultRecord faultRecord = new DryFaultRecord(orderId,tenantId,eqpFault,faultType,new Date(),null);


//                addFauMap.put(redisKey,faultRecord);


                Map<String, DryEquipment> equipmentMap = equipmentService.queryEquByTenantId(tenantId);


                String tenantName = (String) redisUtil.hget(CommonCacheConstant.SYS_CACHE_TENANT, tenantId + "");


                DryFaultRecordVo vo = new DryFaultRecordVo(orderId,tenantId,eqpFault,faultType,new Date(),null,1,equipmentMap.get(machineId).getName(),tenantName);


                addFauMap.put(redisKey,vo);


            }else {


                //如果数据已存在，且计数大于1就重置计数（计数3次后判定故障结束，3次之前重新上报故障说明故障还在持续 需要重新计数）


                if(rFault.getECount()!=null && rFault.getECount() > 1){


                    rFault.setECount(1);


                    redisUtil.hset(MqttConstant.MQTT_REAL_FAULT,redisKey,rFault);


                    System.err.println("报警次数重置 clear clear ，key-"+redisKey);


                }




            }




        }




        //1.3缓存至redis


        //合并数据


        realFauMap.forEach((key, value) -> redFauMap.putIfAbsent(key, value));


        addFauMap.forEach((key, value) -> redFauMap.putIfAbsent(key, value));


        //没有新故障数据不用覆盖


        if(!realFauMap.isEmpty()){


            redisUtil.hmset(MqttConstant.MQTT_EQP_FAULT,redFauMap);


        if(!addFauMap.isEmpty()){


            redisUtil.hmset(MqttConstant.MQTT_REAL_FAULT,redFauMap);


        }








        //2检测已结束的故障


        //2.1如果实时数据不存在redis存在则代表故障结束，存入数据库


        Map<Object, Object> curFauMap = redisUtil.hmget(MqttConstant.MQTT_EQP_FAULT);


        Map<Object, Object> curFauMap = redisUtil.hmget(MqttConstant.MQTT_REAL_FAULT);


        curFauMap.keySet().stream()


                //特别注意，多个报警类型共用方法需要区分类型


                .filter(key -> !realFauMap.containsKey(key) && ((DryFaultRecord)curFauMap.get(key)).getFaultType() == faultType)


                .filter(key -> !realFauMap.containsKey(key) && ((DryFaultRecordVo)curFauMap.get(key)).getFaultType() == faultType)


                .forEach(key -> {


                    System.err.println((faultType == 1 ? "类型：故障" : "类型：报警") +  DateUtils.formatDateTime()+"存入数据库");


                    DryFaultRecord record = (DryFaultRecord)redFauMap.get(key);


                    record.setEndTime(new Date());


                    faultRecordService.save(record);


                    redisUtil.hdel(MqttConstant.MQTT_EQP_FAULT,key);


                    DryFaultRecordVo vo = (DryFaultRecordVo)redFauMap.get(key);


                    vo.setECount(vo.getECount()+1);


                    if(redisUtil.hget(MqttConstant.MQTT_REAL_FAULT,key.toString())!=null){


                        //更新次数


                        redisUtil.hset(MqttConstant.MQTT_REAL_FAULT,key.toString(),vo);


                        System.err.println("报警次数更新，key-"+key);


                    }




                    result.add(record);




                    if(vo.getECount()>=3){


                        vo.setEndTime(new Date());


                        //TODO 结束超过某个时间区间判定为错误数据


                        faultRecordService.save(vo);


                        redisUtil.hdel(MqttConstant.MQTT_REAL_FAULT,key);


                        result.add(vo);


                        System.err.println((faultType == 1 ? "类型：故障" : "类型：报警") +  DateUtils.formatDateTime()+"存入数据库");


                    }


                });








        return result;


    }