elastic-job詳解（一）：數據分片

數據分片的目的在於把一個任務分散到不一樣的機器上運行，既能夠解決單機計算能力上限的問題，也能下降部分任務失敗對總體系統的影響。elastic-job並不直接提供數據處理的功能，框架只會將分片項分配至各個運行中的做業服務器（實際上是Job實例，部署在一臺機器上的多個Job實例也能分片），開發者須要自行處理分片項與真實數據的對應關係。框架也預置了一些分片策略：平均分配算法策略，做業名哈希值奇偶數算法策略，輪轉分片策略。同時也提供了自定義分片策略的接口。

分片原理

elastic-job的分片是經過zookeeper來實現的。分片的分片由主節點分配，以下三種狀況都會觸發主節點上的分片算法執行：

• 新的Job實例加入集羣
• 現有的Job實例下線（若是下線的是leader節點，那麼先選舉而後觸發分片算法的執行）
• 主節點選舉

上述三種狀況，會讓zookeeper上leader節點的sharding節點上多出來一個necessary的臨時節點，主節點每次執行Job前，都會去看一下這個節點，若是有則執行分片算法。

分片的執行結果會存儲在zookeeper上，以下圖，5個分片，每一個分片應該由哪一個Job實例來運行都已經分配好。分配的過程就是上面觸發分片算法以後的操做。分配完成以後，各個Job實例就會在下次執行的時候使用上這個分配結果。

 

每一個job實例任務觸發前都會獲取本任務在本實例上的分片狀況（直接和上圖zookeeper上instance節點比對某一個分片是否該有這個Job實例執行），而後封裝成shardingContext，傳遞給調用任務的實際執行方法：

/**
     * 執行做業.
     *
     * @param shardingContext 分片上下文
     */
void execute(ShardingContext shardingContext);

 

分片算法

全部的分片策略都繼承JobShardingStrategy接口。根據當前註冊到ZK的實例列表和在客戶端配置的分片數量來進行數據分片。最終將每一個Job實例應該得到的分片數字返回出去。 方法簽名以下：

/**
     * 做業分片.
     * 
     * @param jobInstances 全部參與分片的單元列表
     * @param jobName 做業名稱
     * @param shardingTotalCount 分片總數
     * @return 分片結果
     */
    Map<JobInstance, List<Integer>> sharding(List<JobInstance> jobInstances, String jobName, int shardingTotalCount);

分片函數的觸發，只會在leader選舉的時候觸發，也就是說只會在剛啓動和leader節點離開的時候觸發，而且是在leader節點上觸發，而其餘節點不會觸發。

 

1. 基於平均分配算法的分片策略

基於平均分配算法的分片策略對應的類是：AverageAllocationJobShardingStrategy。它是默認的分片策略。它的分片效果以下：

• 若是有3個Job實例, 分紅9片, 則每一個Job實例分到的分片是: 1=[0,1,2], 2=[3,4,5], 3=[6,7,8].
• 若是有3個Job實例, 分紅8片, 則每一個Job實例分到的分片是: 1=[0,1,6], 2=[2,3,7], 3=[4,5].
• 若是有3個Job實例, 分紅10片, 則個Job實例分到的分片是: 1=[0,1,2,9], 2=[3,4,5], 3=[6,7,8].

 

2. 做業名的哈希值奇偶數決定IP升降序算法的分片策略

這個策略的對應的類是：OdevitySortByNameJobShardingStrategy，它內部其實也是使用AverageAllocationJobShardingStrategy實現，只是在傳入的節點實例順序不同，也就是上面接口參數的List<JobInstance>。AverageAllocationJobShardingStrategy的缺點是一旦分片數小於Job實例數，做業將永遠分配至IP地址靠前的Job實例上，致使IP地址靠後的Job實例空閒。而OdevitySortByNameJobShardingStrategy則能夠根據做業名稱從新分配Job實例負載。如：

• 若是有3個Job實例，分紅2片，做業名稱的哈希值爲奇數，則每一個Job實例分到的分片是：1=[0], 2=[1], 3=[]
• 若是有3個Job實例，分紅2片，做業名稱的哈希值爲偶數，則每一個Job實例分到的分片是：3=[0], 2=[1], 1=[]

實現比較簡單：

long jobNameHash = jobName.hashCode();
if (0 == jobNameHash % 2) {
    Collections.reverse(jobInstances);
}
return averageAllocationJobShardingStrategy.sharding(jobInstances, jobName, shardingTotalCount);

 

3. 根據做業名的哈希值對Job實例列表進行輪轉的分片策略

這個策略的對應的類是：RotateServerByNameJobShardingStrategy，和上面介紹的策略同樣，內部一樣是用AverageAllocationJobShardingStrategy實現，也是在傳入的List<JobInstance>列表順序上作文章。

 

4. 自定義分片策略

除了可使用上述分片策略以外，elastic-job還容許自定義分片策略。咱們能夠本身實現JobShardingStrategy接口，而且配置到分片方法上去，整個過程比較簡單，下面僅僅列出經過配置spring來切換自定義的分片算法的例子：

<job:simple id="MyShardingJob1" class="nick.test.elasticjob.MyShardingJob1" registry-center-ref="regCenter" cron="0/10 * * * * ?" sharding-total-count="5" sharding-item-parameters="0=A,1=B,2=C,3=D,4=E" job-sharding-strategy-class="nick.test.elasticjob.MyJobShardingStrategy"/>