2025-12-16 zwy 优化-定时任务

zhaowenyu

2025-12-16 1932bcd3b99877deed7d47acccd2951c1e05e6ce

 ard-sample/src/main/java/com/ard/sample/wayline/service/impl/FlightTaskServiceImpl.java

@@ -140,16 +140,16 @@
                            .build());
        }
        // 进度
        if (eventsReceiver.getOutput().getProgress().getPercent() >= 95) {
            return;
        }
//        if (eventsReceiver.getOutput().getProgress().getPercent() >= 95) {
//            return;
//        }
        String json = "";
        try {
            json = mapper.writeValueAsString(eventsReceiver);
        } catch (Exception e) {
            e.printStackTrace();
        }
        _dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType(), json);
        _dockWebSocketHandler.sendMessageToSubscribers(dockSn, MessageType.STATUS.getType(), json, "");
    }

    /**
@@ -168,40 +168,45 @@
        if (Objects.isNull(jobIdValue)) {
            return;
        }
        log.info("Check the timed tasks of the wayline. {}", jobIdValue);
        log.info("检查 wayline 的定时任务。 {}", jobIdValue);
        // format: {workspace_id}:{dock_sn}:{job_id}
        String[] jobArr = String.valueOf(jobIdValue).split(RedisConst.DELIMITER);
        double time = RedisOpsUtils.zScore(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);
        // 取到待执行的定时任务ID，获取任务信息websocket发给前端
        // region 取定时任务信息websocket发给前端
        // pending status状态待执行
        EventsReceiver<WaylineTaskProgressReceiver> eventsReceiver = new EventsReceiver<>();
        WaylineTaskProgressReceiver waylineTaskProgressReceiver = new WaylineTaskProgressReceiver();
        WaylineTaskProgressExt waylineTaskProgressExt = new WaylineTaskProgressExt();
        waylineTaskProgressExt.setFlightId(jobArr[1]);
        waylineTaskProgressReceiver.setStatus("pending");
        waylineTaskProgressReceiver.setExt(waylineTaskProgressExt);
        eventsReceiver.setOutput(waylineTaskProgressReceiver);
        eventsReceiver.setResult(ResponseResult.CODE_SUCCESS);
        eventsReceiver.setBid(IdUtils.simpleUUID());
        eventsReceiver.setSn(jobArr[0]);
        eventsReceiver.setGateway(jobArr[0]);
        eventsReceiver.setType(MessageType.STATUS.getType() + "-flighttask_progress");
        eventsReceiver.setMethod(EventsMethodEnum.FLIGHT_TASK_PROGRESS.getMethod());
        String json = "";
        try {
            json = mapper.writeValueAsString(eventsReceiver);
        } catch (Exception e) {
            e.printStackTrace();
        // 定时任务执行前，判断有没有任务正在执行，在执行则不发送websocket消息
        String key = RedisConst.WAYLINE_JOB_RUNNING_PREFIX + jobArr[0];
        if (!RedisOpsUtils.checkExist(key)) {
            // 取到待执行的定时任务ID，获取任务信息websocket发给前端
            // pending status状态待执行
            EventsReceiver<WaylineTaskProgressReceiver> eventsReceiver = new EventsReceiver<>();
            WaylineTaskProgressReceiver waylineTaskProgressReceiver = new WaylineTaskProgressReceiver();
            WaylineTaskProgressExt waylineTaskProgressExt = new WaylineTaskProgressExt();
            waylineTaskProgressExt.setFlightId(jobArr[1]);
            waylineTaskProgressReceiver.setStatus("pending");
            waylineTaskProgressReceiver.setExt(waylineTaskProgressExt);
            eventsReceiver.setOutput(waylineTaskProgressReceiver);
            eventsReceiver.setResult(ResponseResult.CODE_SUCCESS);
            eventsReceiver.setBid(IdUtils.simpleUUID());
            eventsReceiver.setSn(jobArr[0]);
            eventsReceiver.setGateway(jobArr[0]);
            eventsReceiver.setType(MessageType.STATUS.getType() + "-flighttask_progress");
            eventsReceiver.setMethod(EventsMethodEnum.FLIGHT_TASK_PROGRESS.getMethod());
            String json = "";
            try {
                json = mapper.writeValueAsString(eventsReceiver);
            } catch (Exception e) {
                e.printStackTrace();
            }
            _dockWebSocketHandler.sendMessageToSubscribers(jobArr[0], MessageType.STATUS.getType(), json, "");
            log.info("定时任务自定义状态为待执行，json: {}", json);
        }
        _dockWebSocketHandler.sendMessageToSubscribers(jobArr[0], MessageType.STATUS.getType(), json);
        log.info("定时任务自定义状态为待执行，json: {}", json);
        // endregion
        long now = System.currentTimeMillis();
        int offset = 30_000;

        // Expired tasks are deleted directly.
        // 过期任务会直接删除。
        if (time < now - offset) {
            log.info("定时任务超时，jobId: {}", jobArr[1]);
            RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);
            waylineJobService.updateJob(WaylineJobDTO.builder()
                    .jobId(jobArr[1])
@@ -214,11 +219,11 @@

        if (now <= time && time <= now + offset) {
            try {
                waylineJobService.executeFlightTask(jobArr[2]);
                waylineJobService.executeFlightTask(jobArr[1]);
            } catch (Exception e) {
                log.info("The scheduled task delivery failed.");
                log.info("计划中的任务交付失败了。jobId: {}，异常信息：{}", jobArr[1], e.getMessage());
                waylineJobService.updateJob(WaylineJobDTO.builder()
                        .jobId(jobArr[2])
                        .jobId(jobArr[1])
                        .status(WaylineJobStatusEnum.FAILED.getVal())
                        .executeTime(LocalDateTime.now())
                        .completedTime(LocalDateTime.now())