XXL Job 是一個開源的分布式任務調度平臺，其核心設計目標是開發迅速、學習簡單、輕量級、易擴展的分布式任務調度框架。

這兩天咱們開發的 AI Cloud 項目中，也使用到了 XXL Job 來執行分布式任務的調度，可以看出它的部署和使用雖然步驟很多，但用起來還是很簡單的。

因為其本身為 Spring Boot 項目，所有對于 Java 程序員來說很友好，而且它還提供中文控制臺，所以這也是他能在國內分布式任務調度系統這塊一直流行的原因，如下圖所示：

那么接下來咱們就來聊聊，XXL Job 的路由策略，以及路由策略中分片任務的執行原理。

1.路由策略

XXL Job 的路由策略主要作用是在任務執行器集群環境中，決定如何選擇合適的執行器來執行任務。

XXL Job 路由策略包含以下幾個：

其中：

• 第一個：選取執行器管理的注冊地址列表中的第一個執行器來執行任務；
• 最后一個：選取執行器管理的注冊地址列表中的最后一個執行器來執行任務；
• 輪詢：依次選取執行器管理的注冊地址列表中的執行器，周而復始。為了應對多個定時任務同時觸發帶來的數據一致性問題，XXL-JOB 使用一個靜態的同步 Map 來存儲每個任務的 jobId 和其對應的計數。每次計數增加后，對執行器地址列表的數量取余，將結果作為索引來獲取對應的執行器地址。如果超過 24 小時沒有觸發調用該任務，會清空 Map 以釋放一定空間；
• 隨機：從執行器管理的注冊地址列表中隨機選取一個執行器來執行任務；
• 一致性 HASH：實現一致性 HASH 負載均衡算法；
• 最不經常使用：選擇最近最少被調度的執行器執行任務（通過次數維度選取任務）；
• 最近最久未使用：選擇距離上次被調度時間最長的執行器執行任務（通過時間維度選取任務），有助于平衡各執行器的工作負載；
• 故障轉移：在任務路由策略選擇“故障轉移”的情況下，如果執行器集群中的某一臺機器出現故障，將會自動 Failover 切換到一臺正常的執行器發送調度請求；
• 忙碌轉移：當任務分配到某個執行器時，如果該執行器正處于忙碌狀態（可能正在執行其他任務或資源緊張），則會嘗試將任務轉移到其他相對空閑的執行器上執行；
• 分片廣播：選取執行器管理的注冊地址列表中的所有地址，每個地址都執行一次任務。這種方式類似于 MQ 的廣播模式，可以將任務廣播到集群中的所有執行器上執行。此策略適用于需要在多個執行器上同時執行相同任務的場景，例如數據同步或分布式計算等。

也就是說在這些路由策略中，最復雜的就是分片廣播了。

2.分片任務實現

所謂的分片廣播也就是分片（執行）任務，它是將一個大任務劃分為多個子任務并行執行，以提高效率。

假設，我們現在要使用分片任務執行一個大數據的查詢與處理，此時的實現代碼如下：

import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.log.XxlJobLogger;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class ShardingBroadcastJob {

    @XxlJob("shardingBroadcastTask") 
    public void execute(String param) {
        // 獲取分片參數：分片總數和分片序列號
        int shardIndex = XxlJobHelper.getShardIndex(); 
        int shardTotal = XxlJobHelper.getShardTotal(); 

        XxlJobLogger.log("當前節點的 index={}, 總結點數={}, 參數={}", shardIndex, shardTotal, param);

        // 模擬獲取數據列表
        List<String> dataList = getDataList(); 

        // 執行分片邏輯
        shardingExecute(dataList, shardIndex, shardTotal);
    }

    public List<String> getDataList() {
        // 這里可以根據實際情況從數據庫或其他數據源獲取數據列表
        // 為了示例簡單，直接返回一個固定的列表
        return List.of("data1", "data2", "data3", "data4", "data5", "data6", "data7", "data8", "data9", "data10");
    }

    public void shardingExecute(List<String> dataList, int shardIndex, int shardTotal) {
        XxlJobLogger.log("開始執行分片任務，當前分片={}, 總分片數={}", shardIndex, shardTotal);

        // 計算當前分片應處理的數據范圍
        int start = (shardIndex * dataList.size()) / shardTotal;
        int end = ((shardIndex + 1) * dataList.size()) / shardTotal;

        // 處理當前分片的數據
        for (int i = start; i < end; i++) {
            String data = dataList.get(i);
            XxlJobLogger.log("處理數據: {}", data);
            // 在此處添加具體的數據處理邏輯
        }

        XxlJobLogger.log("分片任務執行完成");
    }
}

在上述代碼中，在execute方法中，通過 XxlJobHelper.getShardIndex() 獲取當前分片序號，通過 XxlJobHelper.getShardTotal() 獲取總分片數。然后模擬獲取了一個數據列表 dataList，接下來使用 shardingExecute 方法執行分片邏輯。

在 shardingExecute 方法中，根據分片序號和總分片數計算出當前分片應處理的數據范圍，然后遍歷該范圍內的數據并進行處理（此處僅打印數據，實際應用中可添加具體的數據處理邏輯）。

在實際使用時，需要將任務部署到 XXL Job 執行器集群中，并在調度中心配置相應的任務，選擇分片廣播的路由策略。這樣，當調度中心觸發任務時，所有執行器都會執行該任務，并根據分片參數處理相應的數據分片，這樣就能提升程序整體的執行效率了。

3.執行原理

了解了 XXL Job 的代碼實現就能明白其運行原理，它的實現原理如下：

• 任務配置與分發：在 XXL Job 的調度中心，用戶通過 Web 界面創建一個分片廣播類型的任務，并設置相應的參數，如分片總數（shardingTotalCount）。當調度觸發時，調度中心會將此任務廣播至所有注冊的執行器。
• 分片參數傳遞：每個執行器在接收到廣播的任務時，會自動獲得分片參數，包括分片總數和當前執行器應該處理的分片序號（shardingItem）。這些參數由 XXL Job 框架自動注入，使得執行器能夠知道它應當處理哪個數據分片。
• 分片邏輯執行：實際的分片邏輯需要在執行器的任務處理器代碼中實現，開發者需根據分片序號和總數，決定處理哪些數據。這通常涉及對數據源的分片訪問，如數據庫查詢時使用分頁查詢或者 ID 取模等方法來確定每個執行器處理的數據范圍。然后各個執行器并行處理各自分片的數據，互不影響。
• 結果匯總：由于是廣播任務，每個執行器處理的是全量數據的一個子集，因此不存在匯總操作，每個執行器獨立完成自己的處理邏輯。如果需要最終匯總結果，需要額外的邏輯來收集和整合各個執行器的輸出。