点位在线定时任务卡死修复

fuliqi

2024-09-30 13575e2a109e607f3daffcb65103ed0233fa7621

点位在线定时任务卡死修复

 ycl-server/src/main/java/com/ycl/platform/service/impl/WorkOrderServiceImpl.java

@@ -777,7 +777,7 @@
            FFmpegFrameGrabber grabber = FFmpegFrameGrabber.createDefault(rtspUrl);
            if (StringUtils.hasText(rtspUrl)) {
                try {
                    grabber.setOption("rtsp_transport", "tcp"); // 使用tcp的方式，不然会丢包很严重
//                    grabber.setOption("rtsp_transport", "tcp"); // 使用tcp的方式，不然会丢包很严重
                    grabber.setImageWidth(1920);
                    grabber.setImageHeight(1080);
                    grabber.setFrameRate(25);

 ycl-server/src/main/java/com/ycl/task/MonitorTask.java

@@ -80,7 +80,7 @@
    public void synchronize() {
        log.info("开始同步mongodb一机一档到数据库");
        Date yesterday = DateUtils.addDays(new Date(), -1);
        Query query = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(yesterday)).lt(DateUtils.getDayEnd(yesterday)));
        Query query = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date())));
        List<MonitorQualifyResult> oneMachineFileResults = mongoTemplate.find(query, MonitorQualifyResult.class);
        //mongo品牌数据
        Query OSDQuery = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(yesterday)).lt(DateUtils.getDayEnd(yesterday)));

 ycl-server/src/main/java/com/ycl/task/UYTask.java

@@ -3,15 +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.vo.TMonitorVO;
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;
@@ -22,10 +23,10 @@
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;
@@ -33,26 +34,20 @@
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;
@@ -192,17 +187,66 @@
        }
        // 先查出设备IP集合
        List<TMonitorResult> monitorList = monitorMapper.getDistinctIP();
        List<Future<TMonitorResult>> futureList = new ArrayList<>(48);
//        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
//        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;
                            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("部分任务未在指定时间内完成");
        }

        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 -> {
@@ -213,16 +257,14 @@
            return vo;
        }).collect(Collectors.toList());
        monitorMapper.updateOnline(willUpdateList);

        // 工单
        List<WorkOrder> workOrderList = dataList.stream()
                .filter(item -> Objects.nonNull(item.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)) {
            //如果存在之前的数据先删除
@@ -230,7 +272,6 @@
            DeleteResult result = mongoTemplate.remove(query, TMonitorResult.class);
            //存放在mongo中
            mongoTemplate.insertAll(dataList);
            log.info("结束同步点位在线率到mongo");
        }
        log.info("点位在线监测完成");
    }
@@ -320,7 +361,6 @@
//        uyErrorTypeCheckService.recordMetaDSumCheck(mongoTemplate.find(query, RecordMetaDSumResult.class));
        log.info("结束执行录像可用数据同步");
    }



}

 ycl-server/src/main/java/com/ycl/thread/OnlineCheckThread.java

@@ -29,6 +29,7 @@
    private TMonitorResult monitor;
    private CheckPointUtil checkPointUtil;
    private Integer times;

    public OnlineCheckThread(TMonitorResult monitor, CheckPointUtil checkPointUtil,Integer times) {
        this.monitor = monitor;
        this.checkPointUtil = checkPointUtil;
@@ -60,7 +61,7 @@
    }

    @Override
    public TMonitorResult call() throws Exception {
    public TMonitorResult call() {
        TMonitorResult result = checkPointUtil.check(monitor);
        // 一天内监测到离线1次以上，生成工单
        if (result.getOffLineCount() >= times) {
@@ -72,6 +73,7 @@
            workOrder.setStatus(WorkOrderStatusEnum.DISTRIBUTED);
            result.setWorkOrder(workOrder);
        }
        log.info("监测对象："+result);
        return result;
    }


 ycl-server/src/main/java/com/ycl/utils/CheckPointUtil.java

@@ -51,7 +51,6 @@
        } catch (Exception e) {
            monitor.setOnline(Boolean.FALSE);
        }

        // 如果http得到的不在线，那么再ping一下
        boolean reachable = false;
        Integer checkTimes = 1;