Quartz集群增强版_01.集群及缺火处理(ClusterMisfireHandler)
转载请闻名出处 https://www.cnblogs.com/funnyzpc/p/18542452
主要目的
- 应用(app)与节点(node)状态同步
不管是 node 照旧 app,都可以通过对应 state 来控制节点及整个应用的启停,这是很重要的功能,同时对于集群/缺火的锁操作也是基于 app 来做的,同时附加在 app 上的这个锁是控制所有应用及集群之间的并发操作,同样也是很重要的~
- 任务状态与执行状态更新
由于任务扫描主要操作的是执行时间项(execute)信息,同时变更的也是执行项的状态(state),故此需要更新任务(job)状态
- 熄火任务规复执行
任务扫描调度的过程大概存在 GC 及 DB断连 的环境,需要及时修正 next_fire_time 以保证在异通例复后能正常被扫到并被执行
- 清理历史记录
清理的执行频度很低,假如可以的话建议是后管接入 click sdk 手动操作,这里的自动清理是兜底方案,基于数据库锁的任务并发在表数据越少时性能理论上就越好~ ,自动清理有两大任务:
- 1.清理执行无效应用及非执行节点
- 2.清理任务及执行配置
- 创建应用及执行节点
这是必要的操作,预创建节点及应用方便后续管理,同时执行调度也依靠于节点及应用的状态
前置处理
前置处理指的是 Quartz 启动时必做的维护,主要包含三部分主要内容:
- 01.写入应用(app) 及 节点(node) ,这是很重要的
- 02.规复/更新应用状态
- 将执行中或异常的 job 拿出来并检查其关联的执行项,通过执行项(execute)的状态更新任务(job)状态,假如
多执行项存在多个状态,状态的优先级为(从高到低):ERROR->EXECUTING-> AUSED->COMPLETE
代码表象为 :
- List<QrtzExecute> executes = getDelegate().getExecuteByJobId(conn,job.getId());
- boolean hasExecuting = false;
- boolean hasPaused = false;
- boolean hasError = false;
- boolean hasComplete = false;
- for( QrtzExecute execute:executes ){
- final String state = execute.getState();
- if("EXECUTING".equals(state)){
- hasExecuting=true;
- }else if("PAUSED".equals(state)){
- hasPaused=true;
- }else if("ERROR".equals(state)){
- hasError=true;
- }else if("COMPLETE".equals(state)){
- hasComplete=true;
- }else{
- continue; // 这里一般是INIT
- }
- }
- // 如果所有状态都有则按以下优先级来
- String beforeState = job.getState();
- if(hasError){
- job.setState("ERROR");
- }else if(hasExecuting){
- job.setState("EXECUTING");
- }else if(hasPaused){
- job.setState("PAUSED");
- }else if(hasComplete){
- job.setState("COMPLETE");
- }else{
- continue; // 这里对应上面的INIT状态,不做处理
- }
- // 不做无谓的更新...
- if(!job.getState().equals(beforeState)){
- job.setUpdateTime(now);
- getDelegate().updateRecoverJob(conn,job);
- }
复制代码 获取当前应用下的所有执行中或异常的任务(job),并逐步规复任务下所有执行中(EXECUTING)或异常(ERROR)的任务,主要是重新盘算 next_fire_time
后置处理
- 01.后置处理的内容是包含所有前置处理,同时对集群并发做了加锁 (这个很重要,后一段会讲到)
- 02.同步节点状态与应用状态不一致的题目
- 03.更新 check 标记,这个 check 标记主要方便于后续清理之使用,同时 app 上的 check (time_next) 是作为锁定周期的判断依据
?关于并发锁的处理
这个题目可以详细说明一下,一样寻常一个loop(循环)是 15s(TIME_CHECK_INTERVAL) ,在集群环境中同时存在多个节点的并发题目,所以对集群及缺火的处理就存在重复执行
一开始我的思考是按照乐观锁的思路来做,代码大概是这样的:- int ct = getDelegate().updateQrtzAppByApp(conn,app);
- // 5.获取app锁的才可执行 clear 清理以及 recover 恢复,以减少读写
- if( ct>0 ){
- // 获取到锁后的处理
- }
复制代码 但是这样存在重复执行的环境,具体环境先看图:
上图中node1 与 node2 的开始时间相差5s,所以造成了他们获取锁的时间存在5s的时间差别,由于有这5s的存在,多个节点几乎都可以执行这个update语句以获取锁,这样往下的逻辑必然存在重复执行!
任务调度扫描(QuartzSchedulerThread)是同一等到 next_fire_time 的那一刻来竞争锁,而集群/缺火处理(ClusterMisfireHandler)在一个 while 的大循环内 这个循环每次是15s,所以每个节点的所执行的周期是15s(TIME_CHECK_INTERVAL),而锁的竞争却是在执行 update 的那一刻
假如借用 任务扫描(QuartzSchedulerThread )的处理思路就是 再加一个 while 大概 sleep 等候到下一个 check_time(time_next),代码将如下:- long t=0; // 这里的 check_time 就是应用的check时间,loop_time则是当前循环开始时间 if( (t=check_time-loop_time)> 0 ){ Thread.sleep(t); } int ct = getDelegate().updateQrtzAppByApp(conn,app);
- // 5.获取app锁的才可执行 clear 清理以及 recover 恢复,以减少读写
- if( ct>0 ){
- // 获取到锁后的处理
- }
复制代码 以上这样就可以可以基本保证多个node在同一时间竞争同一把锁了... ,这样做另有一个好处,就是基本保证了各个节点的 ClusterMisfireHandler 的循环时间基本一致,同时通过sleep可以随机打散循环时间(添加偏移量)将
ClusterMisfireHandler 的循环处理打散在其他节点执行 。
但是,但是哦,假如使用 sleep + update 的方式 也大概导致同一时间加锁(update)竞争的开销,所以,我借鉴了 shedlock 开源项目的启发,就是思考能不能在竞争锁之前判断锁定时间,获取到锁之后加一个锁定时间
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。 |