blame | 历史 | 补丁 | 提交 | 提交对比 | ignore whitespace
fuliqi
2024-12-09 2587568bec69f9b9956851da73d626c39bc720db 
 ycl-server/src/main/java/com/ycl/task/UYTask.java


@@ -3,6 +3,7 @@


import com.alibaba.fastjson2.JSONObject;


import com.mongodb.client.result.DeleteResult;


import com.ycl.feign.UYClient;


import com.ycl.platform.domain.entity.TMonitor;


import com.ycl.platform.domain.entity.WorkOrder;


import com.ycl.platform.domain.param.UY.ImageDetectionParam;


import com.ycl.platform.domain.param.UY.MonitorQualifyParam;


@@ -15,6 +16,7 @@


import com.ycl.platform.domain.result.UY.VideoOnlineResult;


import com.ycl.platform.domain.vo.UpdateOnlineVO;


import com.ycl.platform.mapper.TMonitorMapper;


import com.ycl.platform.mapper.WorkOrderMapper;


import com.ycl.platform.service.UYErrorTypeCheckService;


import com.ycl.platform.service.WorkOrderService;


import com.ycl.platform.service.YwPointService;


@@ -24,11 +26,14 @@


import com.ycl.utils.CheckPointUtil;


import com.ycl.utils.DateUtils;


import constant.ApiConstants;


import constant.CheckConstants;


import constant.RedisConstant;


import enumeration.ErrorType;


import enumeration.general.WorkOrderStatusEnum;


import lombok.RequiredArgsConstructor;


import lombok.extern.slf4j.Slf4j;


import org.apache.commons.lang3.StringUtils;


import org.springframework.beans.BeanUtils;


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


import org.springframework.data.mongodb.core.MongoTemplate;


import org.springframework.data.mongodb.core.query.Criteria;


@@ -55,6 +60,7 @@


    private final YwPointService pointService;


    private final TMonitorMapper monitorMapper;


    private final WorkOrderService workOrderService;


    private final WorkOrderMapper workOrderMapper;


    private final SysConfigMapper sysConfigMapper;


    private final CheckPointUtil checkPointUtil;


    @Value("${youYun.tenantId}")


@@ -101,7 +107,7 @@


                                item.setNo(item.getDeviceId());


                            }


                        });


                        pointService.setDeviceTagByGB(records);


                        pointService.setDeviceTagByGB(records, CheckConstants.Rule_Category_Video);


                        //存放在mongo中


                        mongoTemplate.insertAll(records);


                        // 工单生成


@@ -150,7 +156,7 @@


                                item.setNo(item.getSerialNumber().getShowValue());


                            }


                        });


                        pointService.setDeviceTagByGB(records);


                        pointService.setDeviceTagByGB(records,CheckConstants.Rule_Category_Or);


                        pointService.setNew(records);


                        //存放在mongo中


                        mongoTemplate.insertAll(records);


@@ -174,8 +180,9 @@




    /**


     * 点位在线PING检测


     * 任务会先执行一次优云同步，然后执行ping检测


     * online字段来自于优云，pingOnline为主动ping检测的。存入mongo给数据中心查阅


     */


    //TODO：点位在线（修改逻辑只针对工单）


    public void pointOnline() throws ExecutionException, InterruptedException {


        log.info("开始检测点位在线");


        Integer times = 2;


@@ -191,11 +198,13 @@


        List<TMonitorResult> monitorList = monitorMapper.getDistinctIP();


        //补充错误时间点


        Query onlineQuery = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date())));


        Map<String, TMonitorResult> mongoMap = mongoTemplate.find(onlineQuery, TMonitorResult.class).stream().collect(Collectors.toMap(TMonitorResult::getNo, Function.identity()));


        Map<String, TMonitorResult> mongoMap = mongoTemplate.find(onlineQuery, TMonitorResult.class)


                .stream()


                .collect(Collectors.toMap(TMonitorResult::getNo, Function.identity(), (existing, replacement) -> replacement));


        for (TMonitorResult result : monitorList) {


            TMonitorResult mongoData = mongoMap.get(result.getNo());


            if(mongoData!=null){


                result.setOffLineTime(mongoData.getOffLineTime());


                result.setOffLineTimeStr(mongoData.getOffLineTimeStr());


            }


        }


        List<TMonitorResult> dataList = new ArrayList<>(48);


@@ -206,37 +215,14 @@


                            return thread.call(); // 假设 OnlineCheckThread 实现了 Callable 接口


                        }, executorService)


                        .orTimeout(60, TimeUnit.SECONDS)


                        //出现异常


                        .exceptionally(ex -> {


                            if (ex instanceof TimeoutException) {


                                log.error("任务执行超时:"+monitor.getIp());


                            } else {


                                log.error("任务执行异常:"+monitor.getIp() + ex);


                                log.error("任务执行异常:"+monitor.getIp() +ex);


                                ex.printStackTrace();


                            }


                            int checkTimes = 1;


                            int offLineTimes = 1;


                            Map<String, Object> map = (Map<String, Object>) redisTemplate.opsForHash().get(RedisConstant.ONLINE_KEY, monitor.getIp());


                            if (!CollectionUtils.isEmpty(map)) {


                                checkTimes = (Integer) map.get("checkTimes") + 1;


                                offLineTimes = (Integer) map.get("offLineTimes");


                            }


                            monitor.setPingOnline(Boolean.FALSE);


                            monitor.setCheckCount(checkTimes);


                            monitor.setOffLineCount(offLineTimes);


                            List<Date> offLineTime = monitor.getOffLineTime();


                            if(CollectionUtils.isEmpty(offLineTime)) offLineTime = new ArrayList<>();


                            offLineTime.add(new Date());


                            monitor.setOffLineTime(offLineTime);


                            if (monitor.getOffLineCount() >= time) {


                                WorkOrder workOrder = new WorkOrder();


                                workOrder.setSerialNumber(monitor.getNo());


                                List<String> errList = new ArrayList<>();


                                errList.add(ErrorType.DEVICE_OFFLINE.getValue());


                                workOrder.setErrorTypeList(errList);


                                workOrder.setStatus(WorkOrderStatusEnum.DISTRIBUTED);


                                monitor.setWorkOrder(workOrder);


                            }


                            return monitor; // 返回失败的结果


                            return null;


                        }))


                .collect(Collectors.toList());




@@ -247,7 +233,7 @@


        try {


            allOf.get(60, TimeUnit.SECONDS); // 给予额外的5秒来收集结果


        } catch (TimeoutException e) {


            log.warn("部分任务未在指定时间内完成");


            log.error("部分任务未在指定时间内完成");


        } catch (Exception e2){


            log.error("数据收集异常"+e2);


        }


@@ -256,27 +242,60 @@


                .map(CompletableFuture::join)


                .filter(Objects::nonNull)


                .collect(Collectors.toList());


        //筛选出ping离线的设备，更改数据库为离线




        Date now = new Date();


        List<UpdateOnlineVO> willUpdateList = dataList.stream().filter(result->!result.getPingOnline()).map(item -> {


            UpdateOnlineVO vo = new UpdateOnlineVO();


            vo.setOnline(ApiConstants.UY_OnlineSite_Offline);


            vo.setIp(item.getIp());


            vo.setUpdateTime(now);


            return vo;


        }).collect(Collectors.toList());


        monitorMapper.updateOnline(willUpdateList);


        List<String> offLineList = new ArrayList<>();


        List<String> onLineList = new ArrayList<>();


        //查出数据库纯车辆或纯人脸设备


//        List<String> serialNumbers = monitorMapper.selectCarOrFace().stream().map(TMonitor::getSerialNumber).collect(Collectors.toList());


        dataList.forEach(item->{


            if(item.getPingOnline()) {


                onLineList.add(item.getIp());


            } else if(!item.getPingOnline()) {


                //筛选出ping离线的设备，更改数据库为离线


                offLineList.add(item.getIp());


            }


        });


        if(!CollectionUtils.isEmpty(offLineList)) {


            monitorMapper.batchUpdateOnline(offLineList, now, ApiConstants.UY_OnlineSite_Offline);


        }


        if(!CollectionUtils.isEmpty(onLineList)) {


            monitorMapper.batchUpdateOnline(onLineList, now, ApiConstants.UY_OnlineSite_Online);


        }


        //存放到mongo


        if (!CollectionUtils.isEmpty(dataList)) {


            List<TMonitorResult> mongoList = new ArrayList<>();


            dataList.forEach(item->{


                String monitorType = item.getMonitorType();


                //同一个设备多个类型每个类型存一条数据，以此在保留数据的情况下区分省厅标签


                if(StringUtils.isNotEmpty(monitorType)){


                    String[] monitors = monitorType.split("/");


                    for (String type : monitors) {


                        TMonitorResult mongoData = new TMonitorResult();


                        BeanUtils.copyProperties(item,mongoData);


                        mongoData.setMonitorType(type);


                        if("1".equals(type)){


                            mongoData.setProvinceTag(mongoData.getProvinceTagVideo());


                        }else if("2".equals(type)){


                            mongoData.setProvinceTag(mongoData.getProvinceTagCar());


                        }else if("3".equals(type)){


                            mongoData.setProvinceTag(mongoData.getProvinceTagFace());


                        }


                        mongoList.add(mongoData);


                    }


                }


            });


            //如果存在之前的数据先删除


            Query query = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date())));


            DeleteResult result = mongoTemplate.remove(query, TMonitorResult.class);


            //存放在mongo中


            mongoTemplate.insertAll(dataList);


            mongoTemplate.insertAll(mongoList);


        }


        //工单（同一IP只生成一个工单）


        //查询数据库已存在的离线工单获取ip集合，剔除


        List<String> ips = workOrderMapper.getOfflineWorkOrder();


        List<WorkOrder> workOrderList = dataList.stream()


                .filter(item -> Objects.nonNull(item.getWorkOrder()))


                .filter(item -> Objects.nonNull(item.getWorkOrder()) && (CollectionUtils.isEmpty(ips) || !ips.contains(item.getIp())))


                .collect(Collectors.toMap(


                        TMonitorResult::getIp,


                        Function.identity(),


@@ -320,7 +339,8 @@


                                item.setNo(item.getDeviceId());


                            }


                        });


                        pointService.setDeviceTagByGB(records);


                        //貌似这里可以忽略省厅标签因为项目展示的后面ping的时候存的数据


                        pointService.setDeviceTagByGB(records,CheckConstants.Rule_Category_Video);


                        //存放在mongo中


                        mongoTemplate.insertAll(records);


                        //更新point表在线状态


@@ -332,7 +352,7 @@


                            vo.setUpdateTime(now);


                            return vo;


                        }).collect(Collectors.toList());


                        monitorMapper.updateOnline(willUpdateList);


                        monitorMapper.updateOnlineFromUyOrHk(willUpdateList);


                        //离线生成工单,一个ip只生成一个工单


                        List<VideoOnlineResult> workOrders = new ArrayList<>(records.stream()


                                .filter(item -> ApiConstants.UY_OnlineSite_Offline.equals(item.getStatus()))


@@ -387,7 +407,7 @@


                        }


                    });


                    //打标签


                    pointService.setDeviceTagByGB(records);


                    pointService.setDeviceTagByGB(records,CheckConstants.Rule_Category_Video);


                    //存放在mongo中


                    mongoTemplate.insertAll(records);


//                    // 工单生成