DolphinScheduler心脏：Quartz的定时任务调度框架深度解析

Quartz是一个开源的Java作业调度框架，它提供了强大的定时任务调度功能。在DolphinScheduler中，Quartz用于实现定时任务的调度和管理。DolphinScheduler通过QuartzExecutorImpl类与Quartz集成，将工作流及其定时管理操作与Quartz调度框架相结合，实现任务的调度执行。
本文将详细剖析Quartz的原理机制，以及在Dolphinscheduler中使用Quartz的原理。
Quartz ER图

• QRTZ_JOB_DETAILS 和 QRTZ_TRIGGERS 是中心表，定义了任务与触发器之间的关系；
  
• QRTZ_TRIGGERS 表通过外键关联了多个触发器类型表，如 QRTZ_SIMPLE_TRIGGERS 和 QRTZ_CRON_TRIGGERS，用于实现不同类型的触发方式；
  
• QRTZ_FIRED_TRIGGERS 用于记录每次任务执行的历史，与任务和触发器表都有关联；
  
• QRTZ_CALENDARS 用于定义触发器的日历排除规则，QRTZ_PAUSED_TRIGGER_GRPS 用于管理触发器组的暂停状态；
  
• QRTZ_SCHEDULER_STATE 和 QRTZ_LOCKS 主要用于集群环境中的任务调度协调，确保高可用性。
  

Dolphinscheduler Quartz使用

新建SHELL任务

流程定义上线并配置调度

定时上线

流程实例运行结果

原理剖析

创建调度

1. org.apache.dolphinscheduler.api.controller.SchedulerController#createSchedule
2. --org.apache.dolphinscheduler.api.service.impl.SchedulerServiceImpl#insertSchedule
3. ....
4. Schedule scheduleObj = new Schedule();
5. Date now = new Date();
6. scheduleObj.setTenantCode(tenantCode);
7. scheduleObj.setProjectName(project.getName());
8. scheduleObj.setProcessDefinitionCode(processDefineCode);
9. scheduleObj.setProcessDefinitionName(processDefinition.getName());
10. ScheduleParam scheduleParam = JSONUtils.parseObject(schedule, ScheduleParam.class);
11. scheduleObj.setCrontab(scheduleParam.getCrontab());
12. scheduleObj.setTimezoneId(scheduleParam.getTimezoneId());
13. scheduleObj.setWarningType(warningType);
14. scheduleObj.setWarningGroupId(warningGroupId);
15. scheduleObj.setFailureStrategy(failureStrategy);
16. scheduleObj.setCreateTime(now);
17. scheduleObj.setUpdateTime(now);
18. scheduleObj.setUserId(loginUser.getId());
19. scheduleObj.setUserName(loginUser.getUserName());
20. scheduleObj.setReleaseState(ReleaseState.OFFLINE);
21. scheduleObj.setProcessInstancePriority(processInstancePriority);
22. scheduleObj.setWorkerGroup(workerGroup);
23. scheduleObj.setEnvironmentCode(environmentCode);
24. scheduleMapper.insert(scheduleObj);
25. ....

复制代码

核心其实就是向 schedule 表中插入了一条数据而已，如下 :

调度上线

1. org.apache.dolphinscheduler.api.controller.SchedulerController#publishScheduleOnline
2. --org.apache.dolphinscheduler.api.service.impl.SchedulerServiceImpl#onlineScheduler
3. ----org.apache.dolphinscheduler.api.service.impl.SchedulerServiceImpl#doOnlineScheduler
4. ------org.apache.dolphinscheduler.scheduler.quartz.QuartzScheduler#insertOrUpdateScheduleTask
6. 精简代码如下 :
7. // TODO 使用schedule id和projectId封装 JobKey，比如jobName=job_25(schedulerId)，jobGroup=jobgroup_1(projectId)
8. JobKey jobKey = QuartzTaskUtils.getJobKey(schedule.getId(), projectId);
9. // TODO 使用projectId和schedule封装jobDataMap，里面封装的是projectId、scheduleId和schedule(JSON存储)
10. Map<String, Object> jobDataMap = QuartzTaskUtils.buildDataMap(projectId, schedule);
11. // TODO 获取cron表达式
12. String cronExpression = schedule.getCrontab();
13. // TODO 获取时区
14. String timezoneId = schedule.getTimezoneId();
16. // TODO 定时调度的开启时间
17. Date startDate = DateUtils.transformTimezoneDate(schedule.getStartTime(), timezoneId);
18. // TODO 定时调度的结束时间
19. Date endDate = DateUtils.transformTimezoneDate(schedule.getEndTime(), timezoneId);
20. jobDetail jobDetail = newJob(ProcessScheduleTask.class).withIdentity(jobKey).build();
21. jobDetail.getJobDataMap().putAll(jobDataMap);
22. // TODO 创建一个Job
23. scheduler.addJob(jobDetail, false, true);
25. // TODO 封装Trigger
26. TriggerKey triggerKey = new TriggerKey(jobKey.getName(), jobKey.getGroup());
27. CronTrigger cronTrigger = newTrigger()
28.                     .withIdentity(triggerKey)
29.                     .startAt(startDate)
30.                     .endAt(endDate)
31.                     .withSchedule(
32.                             cronSchedule(cronExpression)
33.                                     .withMisfireHandlingInstructionIgnoreMisfires()
34.                                     .inTimeZone(DateUtils.getTimezone(timezoneId)))
35.                     .forJob(jobDetail).build();
37. // TODO 开始调度
38. scheduler.scheduleJob(cronTrigger);

复制代码

对应的表

存储每个任务的详细信息

存储触发器的基本信息，是所有触发器类型的父表

存储 Cron 表达式触发器（Cron Trigger）的信息

调度执行

1. org.apache.dolphinscheduler.scheduler.quartz.ProcessScheduleTask，这个类是 qrtz_job_details 中的 JOB_CLASS_NAME 字段
3. protected void executeInternal(JobExecutionContext context) {
4.     JobDataMap dataMap = context.getJobDetail().getJobDataMap();
6.     int projectId = dataMap.getInt(QuartzTaskUtils.PROJECT_ID);
7.     int scheduleId = dataMap.getInt(QuartzTaskUtils.SCHEDULE_ID);
9.     Date scheduledFireTime = context.getScheduledFireTime();
11.     Date fireTime = context.getFireTime();
13.     Command command = new Command();
14.     command.setCommandType(CommandType.SCHEDULER);
15.     command.setExecutorId(schedule.getUserId());
16.     command.setFailureStrategy(schedule.getFailureStrategy());
17.     command.setProcessDefinitionCode(schedule.getProcessDefinitionCode());
18.     command.setScheduleTime(scheduledFireTime);
19.     command.setStartTime(fireTime);
20.     command.setWarningGroupId(schedule.getWarningGroupId());
21.     String workerGroup = StringUtils.isEmpty(schedule.getWorkerGroup()) ? Constants.DEFAULT_WORKER_GROUP
22.             : schedule.getWorkerGroup();
23.     command.setWorkerGroup(workerGroup);
24.     command.setTenantCode(schedule.getTenantCode());
25.     command.setEnvironmentCode(schedule.getEnvironmentCode());
26.     command.setWarningType(schedule.getWarningType());
27.     command.setProcessInstancePriority(schedule.getProcessInstancePriority());
28.     command.setProcessDefinitionVersion(processDefinition.getVersion());
30.     commandService.createCommand(command);
31. }

复制代码

说白了，这个就是quartz的一个回调函数，最终生成Command。
转载自Journey
原文链接：https://segmentfault.com/a/1190000045471756

本文由 白鲸开源 提供发布支持！

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！