executorService.execute() 并发执行 JobExecutor
1. 初始化对象和配置2. 初始化线程池 ExecutorService executorService 用作并发执行 JobExecutor
提交 JobExecutor
CronTrigger 调出触发器封装
- 触发器名称- 任务组- 表达式
- CronTrigger(\"yourTriggerName\
取 executorMap
继承和重写
否
实现接口
1. 初始化对象和配置2. 初始化
(job)JobReRunFromDB
初始化
put DWJob executorJob 队列
调用 execute
类方法
(worker)AbstractJobRunner 抽象类
- synchronized int runJob(DWJob job)- abstract int doRun(DWJob job)- abstract boolean doKill()- boolean kill(Integer jobSchedulerId)
提交 任务到 中用来保存当前正在运行的任务队列
消费 executorJob 队列
(worker)ScriptJobRunner Runner 脚本执行类
- 脚本运行实体类
- DWJob.getCommand 获取需要执行的脚本- Process 运行脚本命令- 记录日志到文件中- 返回脚本执行状态
监控 DB JOB 变化
(worker)Work
- 启动所有服务角色- 初始化线程池 ExecutorService daemonServiceExecutor 用作并发执行各个服务角色
- Executors.newFixedThreadPool(4) 并发实现类- start() 并发执行所有角色- close() 关闭如下服务 - JobExecutorPool 类中的 ExecutorService executorService JobExecutor 并发池 - Worker 类中的 ExecutorService daemonServiceExecutor 角色并发池 - RpcServiceRunner 类中的 rpc 端口服务
(common)SchdulerWorkerFactory
- 调度流程总控制工厂类
SchdulerWorkerFactory 初始化
步骤 1 封装 Job 成 JobExecutor
(quarz)Scheduler 调度服务
- 调度主服务
初始化 executorMap
JobExecutorPool.submit() 任务提交函数
JobDetail 调度信息封装
- 封装调度任务名称- 任务组- 任务实体
- JobDetail(\"testJob\
是
(worker)JobExecutor 线程池任务实体类(实现 Runnable 接口函数)
初始化
删除 Job
(job)JobMonitorFromDB
Job是否任务组
ExecutorService.execute() 并发执行函数
executorService.execute() 方法实时并发执行 slot 中存在的 JobExecutor 任务
(worker)JobRunner 接口
- int runJob(DWJob job) - boolean kill(Integer jobDroneId)
org.quartz.Job 接口
- execute() 执行入口
QuartzScheduler.loadNotScheduleJobs()
(worker)JobSubmitter 提交 JobExecutor 类
- 循环抽取 中 Job 对象- 对满足条件可运行的 job 对象进行 JobExecutor 并根据逻辑规则提交
- 初始化
初始化类
提交 JobExecutor
QuartzScheduler.addschduler() 函数
- 把 DWJob 提交到 addscheduler 函数中做逻辑处理
Quartz 服务组件
步骤 2 调用 submit() 函数 提交 JobExecutor 到 JobExecutorPool 中
(quarz)QuartzScheduler
addschduler() 添加 DWJob 到调度中
1.job 设置为重复提交2. 中删除 Job
QuartzJob 调度任务实体类
- 调度任务实体类- 获取 DWJob 信息- 记录日志等操作
- 实现提交 DWJob 到 executorJob 流程
Scheduler.scheduleJob() 添加任务函数
步骤 3 删除 Job
(worker)JobExecutorPool 线程池类
- 初始化线程池 ExecutorService executorService 用作并发执行 JobExecutor- 对 JobExecutor 运行、并发、生命周期控制
- 实现类 AgentServerHelper.createThreadPool- shutdown 停止 executorService 服务- awaitTerminate 等待终止- submitJob 提交运行 JobExecutor
(quarz)BackUpQuarzData 备份最新 Quarz 数据类
- 备份调度数据到 dw_scheduler_ha_quarz 表中
daemonServiceExecutor.execute() 并发执行如下实例- JobSubmitter 提交 Job 类- JobMonitorFromDB 监控 DB Job 变化类- jobReRunFromDB DB Job 重跑类- RpcServiceRunner RPC thrift 端口
ExecutorService executorService 并发池类
- 实现类 AgentServerHelper.createThreadPool- shutdown() 关闭并发池- execute() 运行 Job
executorService.execute() 实时并发执行 slot 中存在的 JobExecutor 任务
当前 Job 是否正在运行验证 Job.SchedulerId 在 中是否存在
