大疆不同任务类型执行逻辑，上云API源码分析

大疆不同任务类型执行逻辑，上云API源码分析

大疆司空2中有不同的任务类型：立即任务、定时任务、条件任务。
最初我们实现时，选择的是用Quartz创建定时任务，调用API中executeFlightTask接口实现任务下发。
在功能实现之后，随着对API的深入了解，发现大疆API中有相关的任务下发逻辑。
当时只看了实现逻辑，因为活比较多，加上懒，就一直拖到现在。
本文是对源码的学习总结。
所有任务下发的逻辑都从publishFlightTask()方法开始。
com.dji.sample.wayline.service.impl.FlightTaskServiceImpl#publishFlightTask
立即任务

• fillImmediateTime(param)方法，如果是 “立即任务（IMMEDIATE）”，就把任务的执行日期taskDays与执行时间段taskPeriods强制设置为当前服务器时间。
  
• 由于taskDays = [now]，taskPeriods = [ [now] ]，所以只会进入循环一次，并且beginTime = 当前毫秒时间点，endTime = beginTime。
  
• 创建一条飞行任务记录存到数据库中---- waylineJobOpt。
  
• 调用publishOneFlight()方法发布任务
  

[1]publishFlightTask

1.     /**
2.      * 发布飞行任务的核心入口方法。
3.      * 功能：
4.      * 1. 处理立即任务（IMMEDIATE），强制以服务器当前时间作为任务执行时间；
5.      * 2. 根据用户提交的任务日期（taskDays）与时间段（taskPeriods），
6.      *    生成多个 beginTime/endTime 的任务实例；
7.      * 3. 为每个时间段创建一条 WaylineJob（航线任务记录）；
8.      * 4. 若为条件任务，写入对应的触发条件；
9.      * 5. 立即调用 publishOneFlightTask 将任务下发给飞行设备（Dock）；
10.      * 6. 如果任何一次下发失败，则返回失败并终止流程；
11.      * 7. 所有任务下发成功后返回成功响应。
12.      */
13.     @Override
14.     public HttpResultResponse publishFlightTask(CreateJobParam param, CustomClaim customClaim) throws SQLException {
15.         fillImmediateTime(param);
17.         //立即任务只会进入循环一次
18.         for (Long taskDay : param.getTaskDays()) {
19.             LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(taskDay), ZoneId.systemDefault());
20.             for (List<Long> taskPeriod : param.getTaskPeriods()) {
21.                 //立即任务的beginTime = endTime = 当前毫秒时间
22.                 long beginTime = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(0)), ZoneId.systemDefault()))
23.                         .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
24.                 long endTime = taskPeriod.size() > 1 ?
25.                         LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(1)), ZoneId.systemDefault()))
26.                                 .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime;
27.                 //立即任务直接跳过这个判断
28.                 if (TaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) {
29.                     continue;
30.                 }
32.                 //创建一条飞行任务记录
33.                 Optional<WaylineJobDTO> waylineJobOpt = waylineJobService.createWaylineJob(param, customClaim.getWorkspaceId(), customClaim.getUsername(), beginTime, endTime);
34.                 if (waylineJobOpt.isEmpty()) {
35.                     throw new SQLException("Failed to create wayline job.");
36.                 }
37.                 WaylineJobDTO waylineJob = waylineJobOpt.get();
38.                
39.                 //立即任务直接跳过这个方法
40.                 // If it is a conditional task type, add conditions to the job parameters.
41.                 addConditions(waylineJob, param, beginTime, endTime);
43.                 //发布任务
44.                 HttpResultResponse response = this.publishOneFlightTask(waylineJob);
45.                 if (HttpResultResponse.CODE_SUCCESS != response.getCode()) {
46.                     return response;
47.                 }
48.             }
49.         }
50.         return HttpResultResponse.success();
51.     }
52.    
53.     /**
54.      * 如果任务类型为 IMMEDIATE（立即任务），则忽略用户传入的任何日期和时间段，
55.      * 将任务的执行日期（taskDays）与执行时间段（taskPeriods）强制设置为当前服务器时间。
56.      * 立即任务必须由服务器当前时间触发，不允许由客户端自定义时间。
57.      */
58.     private void fillImmediateTime(CreateJobParam param) {
59.         if (TaskTypeEnum.IMMEDIATE != param.getTaskType()) {
60.             return;
61.         }
62.         long now = System.currentTimeMillis() / 1000;
63.         param.setTaskDays(List.of(now));
64.         param.setTaskPeriods(List.of(List.of(now)));
65.     }

复制代码

• deviceRedisService.checkDeviceOnline判断机场是否在线
  
• 调用prepareFlightTask方法，下发飞行准备指令
  
• 调用executeFlightTask方法，下发飞行任务执行指令
  

[2]publishOneFlight

1.     public HttpResultResponse publishOneFlightTask(WaylineJobDTO waylineJob) throws SQLException {
3.         //判断机场是否在线
4.         boolean isOnline = deviceRedisService.checkDeviceOnline(waylineJob.getDockSn());
5.         if (!isOnline) {
6.             throw new RuntimeException("Dock is offline.");
7.         }
9.         //下发飞行准备指令
10.         boolean isSuccess = this.prepareFlightTask(waylineJob);
11.         if (!isSuccess) {
12.             return HttpResultResponse.error("Failed to prepare job.");
13.         }
15.         //下发立即任务执行指令
16.         // Issue an immediate task execution command.
17.         if (TaskTypeEnum.IMMEDIATE == waylineJob.getTaskType()) {
18.             if (!executeFlightTask(waylineJob.getWorkspaceId(), waylineJob.getJobId())) {
19.                 return HttpResultResponse.error("Failed to execute job.");
20.             }
21.         }
23.         ..............(省略部分代码)
25.         return HttpResultResponse.success();
26.     }

复制代码

定时任务

• 遍历定时任务设置中每一天的每一个时间段
  
• 忽略已经过期的时间段
  
• 创建waylineJob并存到数据库中。
  
• 调用publishOneFlightTask()发布任务
  

[1]publishFlightTask

1.     @Override
2.     public HttpResultResponse publishFlightTask(CreateJobParam param, CustomClaim customClaim) throws SQLException {
3.         //定时任务在这个方法中直接返回，不会进行任何处理
4.         fillImmediateTime(param);
6.         //遍历每一天每一个时间段
7.         for (Long taskDay : param.getTaskDays()) {
8.             LocalDate date = LocalDate.ofInstant(Instant.ofEpochSecond(taskDay), ZoneId.systemDefault());
9.             for (List<Long> taskPeriod : param.getTaskPeriods()) {
10.                 long beginTime = LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(0)), ZoneId.systemDefault()))
11.                         .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
12.                 long endTime = taskPeriod.size() > 1 ?
13.                         LocalDateTime.of(date, LocalTime.ofInstant(Instant.ofEpochSecond(taskPeriod.get(1)), ZoneId.systemDefault()))
14.                                 .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() : beginTime;
15.                
16.                 //忽略已经过期的时间段
17.                 if (TaskTypeEnum.IMMEDIATE != param.getTaskType() && endTime < System.currentTimeMillis()) {
18.                     continue;
19.                 }
21.                 //创建waylineJob
22.                 Optional<WaylineJobDTO> waylineJobOpt = waylineJobService.createWaylineJob(param, customClaim.getWorkspaceId(), customClaim.getUsername(), beginTime, endTime);
23.                 if (waylineJobOpt.isEmpty()) {
24.                     throw new SQLException("Failed to create wayline job.");
25.                 }
26.                 WaylineJobDTO waylineJob = waylineJobOpt.get();
27.                
28.                 //不是条件任务，进入这个方法会直接返回。
29.                 // If it is a conditional task type, add conditions to the job parameters.
30.                 addConditions(waylineJob, param, beginTime, endTime);
32.                 //发布任务
33.                 HttpResultResponse response = this.publishOneFlightTask(waylineJob);
34.                 if (HttpResultResponse.CODE_SUCCESS != response.getCode()) {
35.                     return response;
36.                 }
37.             }
38.         }
39.         return HttpResultResponse.success();
40.     }

复制代码

• deviceRedisService.checkDeviceOnline检查机场是否在线。
  
• 调用prepareFlightTask方法，下发飞行准备指令.
  
• 把定时任务添加到Redis的一个Sorted Set（有序集合）里，用任务的开始时间作为排序依据。
  

[2]publishOneFlightTask

1.     public HttpResultResponse publishOneFlightTask(WaylineJobDTO waylineJob) throws SQLException {
3.         //检查机场是否在线
4.         boolean isOnline = deviceRedisService.checkDeviceOnline(waylineJob.getDockSn());
5.         if (!isOnline) {
6.             throw new RuntimeException("Dock is offline.");
7.         }
9.         //下发飞行准备指令
10.         boolean isSuccess = this.prepareFlightTask(waylineJob);
11.         if (!isSuccess) {
12.             return HttpResultResponse.error("Failed to prepare job.");
13.         }
14.         
15.         ..........(省略部分代码)
16.         
17.         //把定时任务添加到Redis有序集合里，用开始时间排序
18.         if (TaskTypeEnum.TIMED == waylineJob.getTaskType()) {
19.             // key: wayline_job_timed, value: {workspace_id}:{dock_sn}:{job_id}
20.             boolean isAdd = RedisOpsUtils.zAdd(RedisConst.WAYLINE_JOB_TIMED_EXECUTE,
21.                     waylineJob.getWorkspaceId() + RedisConst.DELIMITER + waylineJob.getDockSn() + RedisConst.DELIMITER + waylineJob.getJobId(),
22.                     waylineJob.getBeginTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
23.             if (!isAdd) {
24.                 return HttpResultResponse.error("Failed to create scheduled job.");
25.             }
26.         }
28.         return HttpResultResponse.success();
29.     }

复制代码

• checkScheduledJob定时任务，每5s对 Redis 中的定时任务集合扫描一次。
  
• 获取最早的定时任务，拆解任务信息。
  
• 如果任务信息开始比现在早30s (offset)以上，认为任务过期，删除 Redis 队列中的任务，更新任务状态为失败。
  
• 如果任务执行时间在now + offset这个时间区间内，调用executeFlightTask下发执行任务命令，且无论成功或失败，都从 Redis 队列中删除任务信息。
  

[3]checkScheduledJob定时任务
[code]    @Scheduled(initialDelay = 10, fixedRate = 5, timeUnit = TimeUnit.SECONDS)    public void checkScheduledJob() {                //获取最早的定时任务，并拆解任务信息        Object jobIdValue = RedisOpsUtils.zGetMin(RedisConst.WAYLINE_JOB_TIMED_EXECUTE);        if (Objects.isNull(jobIdValue)) {            return;        }        log.info("Check the timed tasks of the 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);        long now = System.currentTimeMillis();        int offset = 30_000;                //任务信息开始比现在早30s以上，认为任务过期，删除Redis队列中的任务，更新任务状态为失败。        // Expired tasks are deleted directly.        if (time < now - offset) {            RedisOpsUtils.zRemove(RedisConst.WAYLINE_JOB_TIMED_EXECUTE, jobIdValue);            waylineJobService.updateJob(WaylineJobDTO.builder()                    .jobId(jobArr[2])                    .status(WaylineJobStatusEnum.FAILED.getVal())                    .executeTime(LocalDateTime.now())                    .completedTime(LocalDateTime.now())                    .code(HttpStatus.SC_REQUEST_TIMEOUT).build());            return;        }        //判断任务执行时间在now + offset这个时间区间内，下发执行任务命令。        if (now