dw_scheduler 架构 - 1.0

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 在 中是否存在