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


@@ -3,13 +3,16 @@


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.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 +21,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 +34,28 @@


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.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}")


@@ -190,18 +186,70 @@


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


        }


        for (Future<OnlineThreadVO> future : futureList) {


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


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


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


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


//        for (TMonitorResult monitor : monitorList) {


//            OnlineCheckThread thread = new OnlineCheckThread(monitor, checkPointUtil, times);


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


//            futureList.add(future);


//        }


//        for (Future<TMonitorResult> future : futureList) {


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


//        }


        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(30, TimeUnit.SECONDS)


                        .exceptionally(ex -> {


                            if (ex instanceof TimeoutException) {


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


                            } else {


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


                            }


                            int checkTimes = 1;


                            int offLineTimes = 1;


                            //TODO:BUG


                            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.setOnline(Boolean.FALSE);


                            monitor.setCheckCount(checkTimes);


                            monitor.setOffLineCount(offLineTimes);


                            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; // 返回失败的结果


                        }))


                .collect(Collectors.toList());




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


        CompletableFuture<Void> allOf = CompletableFuture.allOf(


                futureList.toArray(new CompletableFuture[0])


        );


        try {


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


        } catch (TimeoutException e) {


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


        } catch (Exception e2){


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


        }




        dataList = futureList.stream()


                .map(CompletableFuture::join)


                .filter(result -> result != null)


                .collect(Collectors.toList());


        // 更新point表的在线标识


        Date now = new Date();


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


@@ -212,20 +260,26 @@


            return vo;


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


        monitorMapper.updateOnline(willUpdateList);




        // 工单


        List<WorkOrder> workOrderList = dataList.stream()


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


                .map(OnlineThreadVO::getWorkOrder)


                .map(TMonitorResult::getWorkOrder)


                .collect(Collectors.toList());


        if (CollectionUtils.isEmpty(workOrderList)) {


            return;


        if (!CollectionUtils.isEmpty(workOrderList)) {


            workOrderService.innerAddWorkOrder(workOrderList);


        }


        workOrderService.innerAddWorkOrder(workOrderList);


        //存放到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);


        }


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


    }




    //点位在线率


    //（弃用）点位在线率


    public void videoOnlineTask() {


        //视频图像质量


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


@@ -310,7 +364,6 @@


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


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


    }








}