Quartz集群增強版_01.集群及缺火處理(ClusterMisfireHandler)
Quartz集群增強版_01.集群及缺火處理(ClusterMisfireHandler)
主要目的
-
應用(
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->PAUSED->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);
}
- 03.恢復/更新執行狀態
獲取當前應用下的所有執行中或異常的任務(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 開源項目的啟發，就是思考能不能在競爭鎖之前判斷鎖定時間，獲取到鎖之后加一個鎖定時間??
鎖定時間內的不再去競爭鎖，鎖定時間外的則可以，大致如圖：
看圖，如果我們假定 node1 是先于 node2 執行， 當 node1 在 14:15 成功獲取鎖后 他的下一次執行時間預期就是 14:30 ,同時如果加一個10s的鎖定時間(圖中藍線)，就是在 15:25 及之前是不可以去競爭鎖，這樣當
node2 在 14:20 去嘗試獲取鎖之前發現最近一個鎖定時間點是 14:25 (及之后) ，此時 node2 會自動放棄競爭鎖（執行update），同時進入下一時間點 14:35 并再次判斷鎖定時間點兒，當然這并不是沒有代價的，各位自行領悟吧??
經過改造后的代碼如下:
// TIME_CHECK_INTERVAL 是循環周期，固定為15秒
long tw = TIME_CHECK_INTERVAL/10*3; // 70% 減少并發
if( (app.getTimeNext()-_start)>tw ){
continue;
}
// 5.獲取app鎖的才可執行 clear 清理以及 recover 恢復，以減少讀寫
if( ct>0 ){
// 獲取到鎖后的處理
}
浙公網安備 33010602011771號