blame | 历史 | 补丁 | 提交 | 提交对比 | ignore whitespace
fuliqi
2024-11-18 a1969bc264f5842b87205ffc9fa77e56c51d3058 
 ycl-server/src/main/java/com/ycl/task/UYTask.java


@@ -5,11 +5,15 @@


import com.ycl.feign.UYClient;


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


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


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


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


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


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


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


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


import com.ycl.platform.domain.result.SYS.TMonitorResult;


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


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


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


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


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


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.service.UYErrorTypeCheckService;


@@ -18,11 +22,12 @@


import com.ycl.system.domain.SysConfig;


import com.ycl.system.mapper.SysConfigMapper;


import com.ycl.thread.OnlineCheckThread;


import com.ycl.utils.CheckPointUtil;


import com.ycl.utils.DateUtils;


import com.ycl.utils.http.HttpUtils;


import com.ycl.utils.http.SelfHttpUtil;


import constant.ApiConstants;


import constant.RedisConstant;


import enumeration.ErrorType;


import enumeration.general.WorkOrderStatusEnum;


import lombok.RequiredArgsConstructor;


import lombok.extern.slf4j.Slf4j;


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


@@ -30,36 +35,29 @@


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


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


import org.springframework.data.redis.core.RedisTemplate;


import org.springframework.http.HttpStatus;


import org.springframework.http.HttpStatusCode;


import org.springframework.http.ResponseEntity;


import org.springframework.stereotype.Component;


import org.springframework.util.CollectionUtils;




import java.io.IOException;


import java.net.InetAddress;


import java.text.SimpleDateFormat;


import java.util.*;


import java.util.concurrent.*;


import java.util.function.Function;


import java.util.stream.Collectors;


import java.util.stream.Stream;




//优云对接数据任务


@Slf4j


@Component("UYTask")


@RequiredArgsConstructor


public class UYTask {




    private final RedisTemplate redisTemplate;


    private final MongoTemplate mongoTemplate;


    private final UYClient uyClient;


    private final UYErrorTypeCheckService uyErrorTypeCheckService;


    private final YwPointService pointService;


    private final TMonitorMapper monitorMapper;


    private final SelfHttpUtil selfHttpUtil;


    private final RedisTemplate redisTemplate;


    private final WorkOrderService workOrderService;


    private final SysConfigMapper sysConfigMapper;




    private final CheckPointUtil checkPointUtil;


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


    private String tenantId;


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


@@ -176,7 +174,7 @@


    }




    /**


     * 点位在线检测


     * 点位在线PING检测


     */


    public void pointOnline() throws ExecutionException, InterruptedException {


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


@@ -189,43 +187,130 @@


        } else {


            log.error("请配置离线次数，此次设置为默认值2");


        }


        // 先查出设备IP集合


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


        List<Future<OnlineThreadVO>> futureList = new ArrayList<>(48);


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


        for (TMonitor monitor : monitorList) {


            OnlineCheckThread thread = new OnlineCheckThread(monitor, redisTemplate, selfHttpUtil, times);


            Future<OnlineThreadVO> future = executorService.submit(thread);


            futureList.add(future);


        // 先查出设备IP集合,剔除掉在线情况是未知的,并且只检测正在考核的设备避免多余工单


        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()));


        for (TMonitorResult result : monitorList) {


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


            if(mongoData!=null){


                result.setOffLineTimeStr(mongoData.getOffLineTimeStr());


            }


        }


        for (Future<OnlineThreadVO> future : futureList) {


            dataList.add(future.get()); // get方法会阻塞，知道拿到结果才继续执行for


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


        Integer time = times;


        List<CompletableFuture<TMonitorResult>> futureList = monitorList.stream()


                .map(monitor -> CompletableFuture.supplyAsync(() -> {


                            OnlineCheckThread thread = new OnlineCheckThread(monitor, checkPointUtil, time);


                            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);


                                ex.printStackTrace();


                            }


                            return null;


//                            int checkTimes = 1;


//                            int offLineTimes = 1;


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


//                            if (!CollectionUtils.isEmpty(map)) {


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


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


//                            }else {


//                                map = new HashMap<>();


//                            }


//                            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);


//                            }


//                            map.put("checkTimes", checkTimes);


//                            map.put("offLineTimes", offLineTimes);


//                            redisTemplate.opsForHash().put(RedisConstant.ONLINE_KEY, monitor.getNo(), map);


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




                        }))


                .collect(Collectors.toList());




        // 等待所有任务完成，但不会无限期等待


        CompletableFuture<Void> allOf = CompletableFuture.allOf(


                futureList.toArray(new CompletableFuture[0])


        );


        try {


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


        } catch (TimeoutException e) {


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


        } catch (Exception e2){


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


        }




        // 更新point表的在线标识


        dataList = futureList.stream()


                .map(CompletableFuture::join)


                .filter(Objects::nonNull)


                .collect(Collectors.toList());




        Date now = new Date();


        List<UpdateOnlineVO> willUpdateList = dataList.stream().map(item -> {


            UpdateOnlineVO vo = new UpdateOnlineVO();


            vo.setOnline(item.getOnline());


            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() && !CollectionUtils.isEmpty(serialNumbers) && serialNumbers.contains(item.getNo())) {


                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)) {


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


            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);


        }


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


        List<WorkOrder> workOrderList = dataList.stream()


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


                .map(OnlineThreadVO::getWorkOrder)


                .collect(Collectors.toMap(


                        TMonitorResult::getIp,


                        Function.identity(),


                        (existing, replacement) -> existing // 如果遇到相同的 IP，保留第一个 TMonitorResult 对象


                ))


                .values()


                .stream()


                .map(TMonitorResult::getWorkOrder)


                .collect(Collectors.toList());


        if (CollectionUtils.isEmpty(workOrderList)) {


            return;


        if (!CollectionUtils.isEmpty(workOrderList)) {


            workOrderService.innerAddWorkOrder(workOrderList);


        }


        workOrderService.innerAddWorkOrder(workOrderList);


        log.info("点位在线监测完成");


    }




    //点位在线率


    //点位在线率（优云）


    public void videoOnlineTask() {


        //视频图像质量


        log.info("开始执行点位在线数据同步");


@@ -247,10 +332,34 @@


                        Query query = new Query(Criteria


                                .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date())));


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


                        //打标签


                        records.forEach(item -> {


                            if (Objects.nonNull(item.getDeviceId())) {


                                item.setNo(item.getDeviceId());


                            }


                        });


                        pointService.setDeviceTagByGB(records);


                        //存放在mongo中


                        mongoTemplate.insertAll(records);


                        // 工单生成


                        uyErrorTypeCheckService.videoOnlineCheck(records);


                        //更新point表在线状态


                        Date now = new Date();


                        List<UpdateOnlineVO> willUpdateList = records.stream().map(item -> {


                            UpdateOnlineVO vo = new UpdateOnlineVO();


                            vo.setOnline(item.getStatus());


                            vo.setIp(item.getIpAddr());


                            vo.setUpdateTime(now);


                            return vo;


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


                        monitorMapper.updateOnline(willUpdateList);


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


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


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


                                .collect(Collectors.toMap(


                                        VideoOnlineResult::getIpAddr,


                                        Function.identity(),


                                        (existing, replacement) -> existing // 如果遇到相同的 IP，保留第一个（existing）


                                )).values());


                        uyErrorTypeCheckService.videoOnlineCheck(workOrders);


                    } else {


                        log.error("点位在线结果数据为空{}", data);


                    }


@@ -273,44 +382,44 @@


    public void recordMetaDSumTask() {


        //录像可用率


        log.info("开始执行录像可用数据同步");


//        RecordMetaDSumParam param = new RecordMetaDSumParam();


//        param.setTenantId(tenantId);


//        Calendar instance = Calendar.getInstance();


//        instance.setTime(new Date());


//        instance.add(Calendar.DAY_OF_MONTH, -1);


//        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");


//        String yesterday = format.format(instance.getTime());


//        param.setStatTime(yesterday);


//        JSONObject jsonObject = uyClient.recordMetaDSumList(param);


//        if (jsonObject != null) {


//            if (ApiConstants.UYSuccessCodeStr.equals(jsonObject.getString("code"))) {


//                List<RecordMetaDSumResult> records = jsonObject.getList("data", RecordMetaDSumResult.class);


//                if (!CollectionUtils.isEmpty(records)) {


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


//                    Query query = new Query(Criteria


//                            .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date())));


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


//                    records.stream().forEach(item -> {


//                        if (Objects.nonNull(item.getDeviceId())) {


//                            item.setNo(item.getDeviceId());


//                        }


//                    });


//                    pointService.setDeviceTagByGB(records);


//                    //存放在mongo中


//                    mongoTemplate.insertAll(records);


        RecordMetaDSumParam param = new RecordMetaDSumParam();


        param.setTenantId(tenantId);


        Calendar instance = Calendar.getInstance();


        instance.setTime(new Date());


        instance.add(Calendar.DAY_OF_MONTH, -1);


        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");


        String yesterday = format.format(instance.getTime());


        param.setStatTime(yesterday);


        JSONObject jsonObject = uyClient.recordMetaDSumList(param);


        if (jsonObject != null) {


            if (ApiConstants.UYSuccessCodeStr.equals(jsonObject.getString("code"))) {


                List<RecordMetaDSumResult> records = jsonObject.getList("data", RecordMetaDSumResult.class);


                if (!CollectionUtils.isEmpty(records)) {


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


                    Query query = new Query(Criteria


                            .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date())));


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


                    records.stream().forEach(item -> {


                        if (Objects.nonNull(item.getDeviceId())) {


                            item.setNo(item.getDeviceId());


                        }


                    });


                    //打标签


                    pointService.setDeviceTagByGB(records);


                    //存放在mongo中


                    mongoTemplate.insertAll(records);


//                    // 工单生成


//                    uyErrorTypeCheckService.recordMetaDSumCheck(records);


//                }


//            } else {


//                log.error("录像可用数据为空{}", jsonObject);


//            }


//        }


//         本地测试


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


        uyErrorTypeCheckService.recordMetaDSumCheck(mongoTemplate.find(query, RecordMetaDSumResult.class));


                }


            } else {


                log.error("录像可用数据为空{}", jsonObject);


            }


        }


        // 本地测试


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


//        uyErrorTypeCheckService.recordMetaDSumCheck(mongoTemplate.find(query, RecordMetaDSumResult.class));


        log.info("结束执行录像可用数据同步");


    }








}