redis工单保存故障数据

zhuguifei

3 天以前 f027c87f28485988b3d25f41b0cdf6e5bd42cd42

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

@@ -36,6 +36,8 @@

import java.text.DecimalFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

@Slf4j
@@ -79,6 +81,13 @@

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

    private static final ConcurrentHashMap<String, ReentrantLock> tenantLocks = new ConcurrentHashMap<>();

    private ReentrantLock getLock(String tenantId, String type) {
        String lockKey = tenantId + ":" + type;
        return tenantLocks.computeIfAbsent(lockKey, k -> new ReentrantLock());
    }

    public String getTemporaryToken() {
        if (token == null) {
@@ -171,11 +180,12 @@
            DryOrderVo orderVo = (DryOrderVo) redisUtil.hget(CacheConstants.RedisKeyEnum.WORK_ORDER.getCode(),
                    realTimeDataParentVo.getTenantid() + "_" + realTimeDataParentVo.getMachineid());
            // 1.2 如果有缓存记录
            if (orderVo != null && orderVo.getCode().equals(realTimeDataParentVo.getWorkorder())) {
            if (orderVo != null && orderVo.getCode() != null && orderVo.getCode().equals(realTimeDataParentVo.getWorkorder())) {

                // 1.3 没有缓存记录再查询数据库
            } else {
                // 根据租户id和工单号查询数据库是否有记录，有则返回，没有则新增一条再返回
                realTimeDataVo.setWorkorder(realTimeDataParentVo.getWorkorder());
                orderVo = getOrSaveDryOrderVoDB(realTimeDataVo);
            }
            if (orderVo == null) {
@@ -206,8 +216,12 @@
            orderVo.setOrderStatus(realTimeDataVo.getWorkorder_status());
            orderVo.setEqp_status(realTimeDataVo.getEqp_status());
//        orderVo.setEqp_state(realTimeDataVo.getEqp_state());
            orderVo.setWarning(realTimeDataVo.getEqp_warning());
            orderVo.setFault(realTimeDataVo.getEqp_fault());
            if(realTimeDataParentVo.getFault()!=null && StringUtils.isNotEmpty(realTimeDataParentVo.getFault().getWarning())){
                orderVo.setWarning(realTimeDataParentVo.getFault().getWarning());
            }
            if(realTimeDataParentVo.getFault()!=null && StringUtils.isNotEmpty(realTimeDataParentVo.getFault().getError())){
                orderVo.setFault(realTimeDataParentVo.getFault().getError());
            }
            orderVo.setLevel(realTimeDataVo.getLevel());
            DryOrderTrendVo trendVo = new DryOrderTrendVo(realTimeDataVo);
            // 2.2 保存工单含水率变化 或 重量变化
@@ -226,7 +240,16 @@
        }

        if (realTimeDataParentVo.getFault() != null) {
            fitFaultRecord(realTimeDataParentVo);
            ReentrantLock faultLock = getLock(realTimeDataParentVo.getTenantid() + "", "fault");
            faultLock.lock();
            try {
                fitFaultRecord(realTimeDataParentVo);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                faultLock.unlock();
            }

        }
        return Result.ok();
    }
@@ -291,7 +314,7 @@
     * @return
     */
    private DryOrderVo getOrSaveDryOrderVoDB(RealTimeDataVo realTimeDataVo) {
        TenantContext.setTenant(realTimeDataVo.getTenantid() +"");
        TenantContext.setTenant(realTimeDataVo.getTenantid() + "");
        DryOrderVo orderVo;
        LambdaQueryWrapper<DryOrder> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(DryOrder::getCode, realTimeDataVo.getWorkorder());
@@ -329,7 +352,7 @@
     * @return
     */
    private DryOrderVo saveNewOrder(RealTimeDataVo realTimeDataVo) {
        TenantContext.setTenant(realTimeDataVo.getTenantid() +"");
        TenantContext.setTenant(realTimeDataVo.getTenantid() + "");
        DryOrderVo orderVo;

        // 查询设备
@@ -357,16 +380,26 @@
                    new LambdaQueryWrapper<DryEqpType>()
                            .eq(DryEqpType::getTenantId, realTimeDataVo.getTenantid())
            );
            if(eqpType == null){
                log.error("未查询到租户设备类型：{}", realTimeDataVo.getTenantid() );
            if (eqpType == null) {
                log.error("未查询到租户设备类型：{}", realTimeDataVo.getTenantid());
                return null;
            }

            Optional.ofNullable(eqpType).ifPresent(type -> addEqu.setType(type.getId()));

            if (!equipmentService.save(addEqu)) {
                log.error("新增设备失败：数据库保存异常！equipment={}", addEqu);
                return null;
            // 设备新增
            ReentrantLock equipmentLock = getLock(realTimeDataVo.getTenantid() + "", "equipment");

            equipmentLock.lock();
            try {
                if (!equipmentService.save(addEqu)) {
                    log.error("新增设备失败：数据库保存异常！equipment={}", addEqu);
                    return null;
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                equipmentLock.unlock();
            }
            equ = addEqu;

@@ -380,14 +413,26 @@
            log.error("未找到药材：" + realTimeDataVo.getIndex() + "," + realTimeDataVo.getName() + ",机台：" + realTimeDataVo.getMachineid());
            return null;
        }
        // 创建新工单
        orderVo = new DryOrderVo(realTimeDataVo);

        orderVo.setHerbId(herbFormula.getId());
        orderVo.setEquId(equ.getId());
        DryOrder dryOrder = BeanUtil.toBean(orderVo, DryOrder.class);
        boolean save = dryOrderService.save(dryOrder);
        return orderVo;
        // 工单新增
        ReentrantLock orderLock = getLock(realTimeDataVo.getTenantid() + "", "order");
        orderLock.lock();
        try {
            // 创建新工单
            orderVo = new DryOrderVo(realTimeDataVo);
            orderVo.setHerbId(herbFormula.getId());
            orderVo.setEquId(equ.getId());
            DryOrder dryOrder = BeanUtil.toBean(orderVo, DryOrder.class);
            boolean save = dryOrderService.save(dryOrder);
            return orderVo;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            orderLock.unlock();
        }


        return null;
    }


@@ -414,9 +459,9 @@
                object.put("tenantId", realTimeDataVo.getTenantid());
                mqttMessage.setPayload(object.toJSONString().getBytes());
                try {
                   if(mqttEnable){
                       mqttUtil.getMqttClient().publish(MqttConstant.SERVICE_REQ_PREFIX, mqttMessage);
                   }
                    if (mqttEnable) {
                        mqttUtil.getMqttClient().publish(MqttConstant.SERVICE_REQ_PREFIX, mqttMessage);
                    }
                } catch (MqttException e) {
                    e.printStackTrace();
                }
@@ -441,7 +486,7 @@
        if (one == null) {
            one = new DryHerbFormula(realTimeDataVo);
            DryEquipment dryEquipment = equipmentService.selectByTenantIdEquipmentId(realTimeDataVo.getTenantid() + "", realTimeDataVo.getMachineid());
            if (dryEquipment!=null&&dryEquipment.getType()!=null) {
            if (dryEquipment != null && dryEquipment.getType() != null) {
                one.setEqpType(dryEquipment.getType());
            }

@@ -507,7 +552,7 @@
                    // 获取工单
                    DryOrderVo order = (DryOrderVo) redisUtil.hget(CacheConstants.RedisKeyEnum.WORK_ORDER.getCode(), realTimeDataVo.getTenantid() + "_" + item.getCode());
                    list.add(item.getName().substring(0, item.getName().indexOf('#') + 1));
                    if (order != null) {
                    if (order != null &&  order.getDetailList()!=null && !order.getDetailList().isEmpty()) {
                        // 计算干燥效率，用于对比
                        DryOrderTrendVo dryOrderTrendVo = order.getDetailList().get(order.getDetailList().size() - 1);
                        double v = order.getOriginWeight() - dryOrderTrendVo.getWeight();
@@ -657,51 +702,44 @@
    @Override
    public void fitFaultRecord(RealTimeDataParentVo vo) {
        TenantContext.setTenant(vo.getTenantid() + "");
        ThreadUtil.execute(() -> {
            try {
                //解析存储报警数据
                List<DryFaultRecord> errorList = fitFault(vo.getFault().getError(), vo.getWorkorder(), vo.getTenantid(), vo.getMachineid(), 1);
                List<DryFaultRecord> warnList = fitFault(vo.getFault().getWarning(), vo.getWorkorder(), vo.getTenantid(), vo.getMachineid(), 2);
                if(!errorList.isEmpty()){
                   log.error("保存故障：{}", errorList.toString());
                }
                if(!warnList.isEmpty()){
                    log.error("保存告警：{}", warnList.toString());
                }
        //解析存储报警数据
        List<DryFaultRecord> errorList = fitFault(vo.getFault().getError(), vo.getWorkorder(), vo.getTenantid(), vo.getMachineid(), 1);
        List<DryFaultRecord> warnList = fitFault(vo.getFault().getWarning(), vo.getWorkorder(), vo.getTenantid(), vo.getMachineid(), 2);
        if (!errorList.isEmpty()) {
            log.error("保存故障：{}", errorList.toString());
        }
        if (!warnList.isEmpty()) {
            log.error("保存告警：{}", warnList.toString());
        }

                //以下为云服务处理故障,厂内本地服务无需处理
                if(!mqttEnable)return;
        //以下为云服务处理故障,厂内本地服务无需处理
        if (!mqttEnable) return;



                //处理结束后，将redis中实时数据发送至云服务器  key = tenantId + machineId + eqpFault
                Map<Object, Object> toCloudFaultMap = redisUtil.hmget(String.format(MqttConstant.MQTT_REAL_FAULT, vo.getTenantid()));
        //处理结束后，将redis中实时数据发送至云服务器  key = tenantId + machineId + eqpFault
        Map<Object, Object> toCloudFaultMap = redisUtil.hmget(String.format(MqttConstant.MQTT_REAL_FAULT, vo.getTenantid()));


                Map<String, DryFaultRecordVo> dryFaultMap = toCloudFaultMap.entrySet().stream()
                        .collect(Collectors.toMap(
                                entry -> entry.getKey().toString(),
                                entry -> (DryFaultRecordVo)entry.getValue()
                        ));
        Map<String, DryFaultRecordVo> dryFaultMap = toCloudFaultMap.entrySet().stream()
                .collect(Collectors.toMap(
                        entry -> entry.getKey().toString(),
                        entry -> (DryFaultRecordVo) entry.getValue()
                ));

                String tenantId = vo.getTenantid() +"";
        String tenantId = vo.getTenantid() + "";

                //广播发送给各租户下移动设备
                if (dryFaultMap.isEmpty()) {
                    return;
                }
                String recTopic = String.format(MqttConstant.SERVICE_BROADCAST_TENANT_REAL_FAULT, tenantId);
                //数据转换
                List<DryFaultRecordVo> faultList = new ArrayList<DryFaultRecordVo>((dryFaultMap.values()));
                MqMessage<List<DryFaultRecordVo>> mqMessage = new MqMessage<>(faultList, tenantId, recTopic);
                //发送广播
                log.error("广播给：{}" , recTopic);
                mqttUtil.sendMqttMessage(MqttConstant.SERVICE_BROADCAST_TENANT_REAL_FAULT, mqMessage, 1);
        //广播发送给各租户下移动设备
        if (dryFaultMap.isEmpty()) {
            return;
        }
        String recTopic = String.format(MqttConstant.SERVICE_BROADCAST_TENANT_REAL_FAULT, tenantId);
        //数据转换
        List<DryFaultRecordVo> faultList = new ArrayList<DryFaultRecordVo>((dryFaultMap.values()));
        MqMessage<List<DryFaultRecordVo>> mqMessage = new MqMessage<>(faultList, tenantId, recTopic);
        //发送广播
        log.error("广播给：{}", recTopic);
        mqttUtil.sendMqttMessage(MqttConstant.SERVICE_BROADCAST_TENANT_REAL_FAULT, mqMessage, 1);

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

    }

@@ -759,12 +797,11 @@
//                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);
                DryFaultRecordVo vo = new DryFaultRecordVo(orderId, tenantId, eqpFault, faultType, new Date(), null, 1, machineId, equipmentMap.get(machineId).getName(), tenantName);
                addFauMap.put(redisKey, vo);
            } else {
                //TODO 特殊情况，如果redis的故障和新
                
                


                //如果数据已存在，且计数大于1就重置计数（计数3次后判定故障结束，3次之前重新上报故障说明故障还在持续 需要重新计数）
                if (rFault.getECount() != null && rFault.getECount() > 1) {
                    rFault.setECount(1);
@@ -787,27 +824,33 @@
        //2检测已结束的故障
        //2.1如果实时数据不存在redis存在则代表故障结束，存入数据库
        Map<Object, Object> curFauMap = redisUtil.hmget(String.format(MqttConstant.MQTT_REAL_FAULT, tenantId));
        curFauMap.keySet().stream()
        Map<Object, DryFaultRecordVo> collect = curFauMap.keySet().stream()
                .filter(key -> {
                    String[] split = key.toString().split("_");
                    return split[0].equals(tenantId + "") && split[1].equals(machineId);
                })
                //特别注意，多个报警类型共用方法需要区分类型
                .filter(key -> !realFauMap.containsKey(key) && ((DryFaultRecordVo) curFauMap.get(key)).getFaultType() == faultType)
                .forEach(key -> {
                    DryFaultRecordVo vo = (DryFaultRecordVo) redFauMap.get(key);
                    vo.setECount(vo.getECount() + 1);
                    if (redisUtil.hget(String.format(MqttConstant.MQTT_REAL_FAULT, tenantId), key.toString()) != null) {
                        //更新次数
                        redisUtil.hset(String.format(MqttConstant.MQTT_REAL_FAULT, tenantId), key.toString(), vo);
                        System.err.println("报警次数更新，key-" + key);
                    }
                .filter(key -> !realFauMap.containsKey(key) && (Objects.equals(((DryFaultRecordVo) curFauMap.get(key)).getFaultType(), faultType)))
                .collect(Collectors.toMap(key -> key,  // 保留原键
                        key -> (DryFaultRecordVo) curFauMap.get(key)));
        collect.keySet().forEach(key -> {
            DryFaultRecordVo vo = (DryFaultRecordVo) redFauMap.get(key);
            vo.setECount(vo.getECount() + 1);
            if (redisUtil.hget(String.format(MqttConstant.MQTT_REAL_FAULT, tenantId), key.toString()) != null) {
                //更新次数
                redisUtil.hset(String.format(MqttConstant.MQTT_REAL_FAULT, tenantId), key.toString(), vo);
                System.err.println("报警次数更新，key-" + key);
            }

                    if (vo.getECount() >= 3) {
                        vo.setEndTime(new Date());
                        //TODO 结束超过某个时间区间判定为错误数据
                        faultRecordService.save(vo);
                        redisUtil.hdel(String.format(MqttConstant.MQTT_REAL_FAULT, tenantId), key);
                        result.add(vo);
                        System.err.println((faultType == 1 ? "类型：故障" : "类型：报警") + DateUtils.formatDateTime() + "存入数据库");
                    }
                });
            if (vo.getECount() >= 3) {
                vo.setEndTime(new Date());
                //TODO 结束超过某个时间区间判定为错误数据
                faultRecordService.save(vo);
                redisUtil.hdel(String.format(MqttConstant.MQTT_REAL_FAULT, tenantId), key);
                result.add(vo);
                System.err.println((faultType == 1 ? "类型：故障" : "类型：报警") + DateUtils.formatDateTime() + "存入数据库");
            }
        });


        return result;