?任務編排工具在之家業務系統中應用探究

本文主要介紹在之家廣告業務系統中運用任務編排治理工具的場景及其可以解決的問題，講解任務編排框架的核心要點，以及向大家演示一個任務編排框架的基本結構，讓大家對任務編排工具增強業務開發效率，提高研發質量有更深刻的理解，

我們開始以下面的實際業務場景代入任務編排工具的使用，

廣告引擎流量接入層中一個用戶流量請求過來，系統會對流量進行一系列的驗證，資料加工處理最侄訓將用戶流量轉化為廣告引擎可召回的廣告請求，具體程序如下：

從圖 1 左側圖中可以看到，實際一個廣告流量在真正發起廣告請求時會經歷反作弊，請求染色， 搜索分詞，IP分詞，用戶畫像等5個任務步驟的層層加工處理，同時第一個任務是前置條件，后面四個任務程序是可以并行處理，任務間彼此無依賴，最終這五個步驟完畢后會統一進行任務結果的聚合，也就是達到異步請求聚合的節點，

上面的任務流翻譯成任務節點順序就變成了上圖 1 中右側的內容了，針對這種一個事情有多個任務可以并發執行并最侄訓聚結果的情況，是很典型的異步并發編程實踐;編程實作中可以很方便使用 Java 語言 JUC 中自定義執行緒池 + CompletableFuture + 執行緒計數器CountDownLatch 來并發執行任務，計數器等待所有任務完成后匯總結果，

但如果我們業務場景再變復雜一些，比如拿部門電商平臺商品詳情頁面展示介面資料匯聚程序舉例，商詳頁面需要分別從商品產品側，商品類別側獲取資料，而每一側的資料又需要劃分成不同的任務分別向不同的地方來獲取資料，最后將獲得資料的分類聚合的業務場景，我用一個抽象圖描述下這個任務程序圖如下：

圖2

這個任務拆解圖與上個場景中的內容來看最大的區別是多了一組異步并發任務，如果繼續采用原編程方法，具體的程序代碼結構簡潔示例如下：

// 并行處理任務 Product 的任務 (例如：TaskA 下的任務串列)
        @Resource
        List<ProductTask> productTasks;
    
        // 依賴于Item的 任務 (例如：TaskB 下的任務串列)
        @Resource
        List<ItemTask> itemTasks;
    
        public void testFuture(HttpServletRequest httpServletRequest) {
            DataContext dataContext = new DataContext();
            dataContext.setHttpServletRequest(httpServletRequest);
           // 第一個并發編程控制環節
            List<Future> product = new ArrayList<>();
            for (Task asyncTask : productTasks) {
                Future<?> submit = threadPoolExecutor.submit(() -> {
                    asyncTask.task(dataContext, null);
                });
                product.add(submit);
            }
            for (Future future : product) {
                try {
                    future.get();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
    
           // 第二個并發編程控制環節
            List<Future> item = new ArrayList<>();
            for (Task asyncTask : itemTasks) {
                Future<?> submit = threadPoolExecutor.submit(() -> {
                    asyncTask.task(dataContext, null);
                });
                item.add(submit);
            }
            for (Future future : item) {
                try {
                    future.get();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
           
           // 當前 2 個環節都執行完成后，下面可以進行的結果匯總程序
        }

●存在兩個多執行緒異步并發編程代碼相似的處理流程，如果業務場景再復雜一些，比如有 3 個或是 4 個并發場景呢？或是某一個并發場景中存在任務上下依賴的情況，可以想象在這個程序中，運行多執行緒編程的代碼也會變成特別臃腫，多執行緒處理那套模板會出現多次，且無法清晰地看清任務的流程，編程人員針對這種復雜的業務流程要處理兩大類的事務，一類是多執行緒技術運用對任務流程的控制，一類是業務本身的邏輯實作，

●兩個多執行緒異步并發編程代碼段，比如 TaskA, TaskB 包含的子任務都實作了多執行緒異步并發編程，但 TaskA 與 TaskB 之間是無法并發編程的，需要 TaskA 那段完成之后，才能順序執行 TaskB 那段，也就是 TaskB 這一整塊需要等待 TaskA 整個任務的完成，但實際組裝整個商詳頁面資料，這兩個任務間是可以不用彼此等待的， 所以會產生等待時間影響系統整體 QPS，

所以遇到業務復雜的場景，我們會考慮采用分而治之的思想把業務拆解成一個個子任務塊來執行，并會使用多執行緒技術來提效實作整個業務流轉，此時多執行緒編程技術的運用也會變得復雜，在實際任務間存在依賴關系，并發限制，超時和錯誤處理，多階段任務等情況需要一并解決的場景下，單純地采用傳統的多執行緒編程技術（CompletableFuture）會讓編程體驗下降很多，任務間，資料間的依賴并不清晰，后續維護時需要針對業務中的任務塊進行更換、順序調整或者由串行改并行時需要增加很多重構作業，

不同業務場景中的任務拆分之后，其組合的的場景也是豐富的，很多時候是串行&并行并存的場景，比如如下幾個例子：

圖3

所以基于上述痛點分析后，如果能有一種多執行緒編程技術封裝，能將多個任務按照依賴關系和執行順序組合起來的技術，這樣業務編程人員在實際業務實作中能更加專注于業務本身，無需關注多執行緒代碼結構的維護，任務執行的細節，而且還能提高開發效率和業務系統的可維護性，那將會對研發體驗與研發質量都將有質的提升，

而這就是任務編排技術的范疇，

任務編排是一種將多個任務按照依賴關系和執行順序組合起來的技術，可以把"任務"這個原子單位按照自己的方式進行編排，任務之間可能互相依賴，復雜一點的編排之后就能形成一個作業流，

在業務系統開發中，一些業務復雜的場景需要同時處理多個任務，這些任務之間可能有先后順序和依賴關系，此時通過任務編排技術可以很方便地組合這些任務，滿足業務需求，

與傳統多執行緒編程相比，任務編排的最大優點是易于調度和控制，因為任務編排框架通常會提供一組方法介面來處理任務之間的依賴關系和執行順序，編程人員只需要根據任務的性質和優先級來描述整個任務流程，簡化了管理和調度的復雜性，

相反，對于只利用多執行緒去處理任務的方式，程式擁有復雜的執行緒控制邏輯，修改各個執行緒共享資料的方法，有大量重復的多執行緒編程代碼，這將增加編程難度和程式的不穩定性，

此外，使用任務編排的模型，可以根據任務流程進行系統分析，提出資源高峰并進行優化，從而提高系統效率，

總之，這些業務開發場景都需要用到任務編排，使用任務編排可以使得業務開發人員更加專注于業務本身，無需關注任務執行的細節，提高開發效率，改善編程體驗以及增強業務系統的可維護性，

以下是我列舉要實作一個任務編排框架需要思考的問題：

1.   任務間的依賴關系如何維護，用什么方式可以存盤這些原子任務間的彼此關系？

2.   任務間的資訊如何傳遞，任務引數如何透傳，任務結果如何傳遞給下一個任務？

3.   任務流程始何跑起來，如何起到控制作用；先執行什么，后執行什么，如何實作任務間并行執行，又是如何控制并發執行任務的前后順序的？

4.   任務運行中出現了例外怎么感知，怎么處理，會對整個流程有什么影響？

下面我們針對這4類問問整體進行一個個拆解，并配以流程圖及示例代碼來講解具體的框架設計，以及簡潔明了的解釋為何這樣設計;在開始介紹前我們先有以下的前提

●前提一：該任務編排框架定位于業務系統開發程序中，用于替代傳統多執行緒編程流程控制那個環節，目標是應用框架工具后，業務開發人員更加專注于業務本身，無需關注任務并行執行，任務流程控制的細節，這些場景有別于我們熟知且常見的大資料分析平臺中的任務編排場景，

●前提二：我們已經對業務系統的原子性任務己經有了完善合理的拆分，并成功的梳理出任務間的上下依賴關系，

●前提三：這將是一個小而精的輕量級任務編排，任務并發執行的流程控制框架，你可以理解其就是對常用 JUC 工具包的二次包裝，是可以提取成一種工具類，也可以抽象成一種任務編排并發執行平臺，后續內容將向大家講解流程編排這個環節的核心設計，

圖4

整個框架需要的角色相對較少，主要包含任務啟動器，任務流程決議器，任務流程處理器，任務總線（TaskContext）以及業務任務塊組成，

任務啟動器扮演入口角色，負責整個流程的啟動，是與其它系統模塊互動的門戶，

任務流程決議器記錄著所有業務任務塊間的前后依賴關系，可以理解是一堆配置，這部分配置資訊可以由多種資料容器來承載，

任務流程處理器是這個任務編排的心臟，其負責從任務流程決議器中解讀出編排好的任務間關系，并統一采用多執行緒并發編程技術來并發執行這些任務，需要對任務進行前后執行順序進行控制，并需要對在運行中任務出現了例外進行感知，

業務任務塊中是真正業務功能實作者，其宣告多個回呼方法讓任務流程處理器在適當的時機進行呼叫，從而來完成業務需求的功能，最終達到業務需求目錄，

任務總線會貫穿整體任務編排并發執行程序的始終，用于傳遞任務啟動時傳入的任務引數，任務配置等資訊，

當所有任務執行完畢后，所有任務的執行結果會匯總，用于最終的業務回傳，

任務編排的啟動者，啟動類 TaskScheduling 是業務方呼叫某一個任務編排的統一入口，其負責后續的任務流的啟動，

圖5

該啟動入口主要是給業務方指定任務啟動時的環境引數，該類是整體框架的門面，整個任務編排程序全部被其封裝在內部，通過提供要求的輸入引數來完成編排好的任務啟動，具體該類的宣告示例代碼如下：

public class TaskScheduling {
         /**
         * 用戶自定義執行緒池
         */
        private static ExecutorService executorService;
       
       /**
         * 任務編排啟動入口
         * @param taskContext       任務執行背景關系物件，存放的有任務引數
         * @param executorService   用戶指定的執行緒池
         * @param nodes             編排好任務節點串列，啟動時可以只傳入頂節點
         * @param timeout           任務編排整體超時時長限制
         * @return                  回傳每一個任務的結果
         */
        public static Map<String, SchedulingNode> start(TaskContext taskContext, ExecutorService executorService, List<SchedulingNode> nodes,long timeout) {
            //定義一個map，存放所有的node，key為node唯一name，value是該node，流程結束之后可以從value中獲取node的結果
            ConcurrentHashMap<String, SchedulingNode> resultNode = new ConcurrentHashMap<>();
            //執行緒池確認
            TaskScheduling.executorService = executorService;
           //編排好的流程正式啟動
            CompletableFuture[] count = new CompletableFuture[nodes.size()];
            for (int i = 0; i < nodes.size(); i++) {
                SchedulingNode node = nodes.get(i);
                count[i] = CompletableFuture.runAsync(()->{
                node.execute(executorService,node, timeout, resultNode, taskContext);
            },executorService);
            }
           // 等待編排好的流程全部結束
            try {
                CompletableFuture.allOf(count).get(timeout, TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                // 例外的一些處理
            }
            return resultNode;
        }

任務流程決議器模塊是整體任務編排框架重要的的模塊，主要有2大職責，分別是：

●按照某種資料結構提供資料容器來保存每一個被編排的任務，并能維護好任務間彼此的關系

●提供任務關系決議引擎，方便后續流程讀取任務依賴的規則

能承載任務編排作業流資訊的資料容器簡單可以使用父子兩個串列來實作，也可以使用較通用的 DAG 有向無環圖這種資料結構，這里簡單提下 DAG ,具體的代碼落地存盤圖可以使用鄰接矩陣和鄰接表這兩種資料結構，這兩種示例圖如下：

圖6

以鄰接矩陣資料結構為例：比如節點1之后有 2 個子節點分別是節點 2 和節點 4 那么可以采用鄰接矩陣中 Array[1][2]，  Array[1][4]標識為 1 即可，當整個矩陣的點按有向無環關系標記完成后，這個矩陣內的二維空間值就維護出了任務節點間的依賴，這樣一組任務編排關系就可以落地了，

此外我們也可以使用鄰接表來存盤，這種存盤方式較好地彌補了鄰接矩陣浪費空間的缺點，但鄰接矩陣能更快地判斷連通性，所以代碼實作上，我們會選擇鄰接矩陣，這樣在判斷兩點之間是否有邊更方便點，具體 Java 代碼實作推薦大家看下開源工程 powerJob 中 DAG作業流的代碼，

在初始實作一個任務編排框架時，往往會先使用簡單易理解的容器來實作,下面我向大家示例采用父子兩個串列的方式來承載任務編排作業流資訊，

/**
         * 依賴的父節點
         */
        protected List<SchedulingNode> fatherHandler = new ArrayList<>();
        /**
         * 系結下游的子節點
         */
        protected List<SchedulingNode> sonHandler = new ArrayList<>();
    
      public SchedulingNode<T, V> setSonHandler(List<SchedulingNode> nodes){
            this.sonHandler = nodes;
            return this;
        }
        public SchedulingNode<T, V> setFatherHandler(List<SchedulingNode> nodes){
            this.fatherHandler = nodes;
            return this;
        }
        public SchedulingNode<T, V> setSonHandler(SchedulingNode...nodes){
            ArrayList<SchedulingNode> nodeList = new ArrayList<>();
            for (SchedulingNode node : nodes){
                nodeList.add(node);
            }
            return setSonHandler(nodeList);
        }
        public void setFatherHandler(SchedulingNode...nodes){
            ArrayList<SchedulingNode> nodeList = new ArrayList<>();
            for (SchedulingNode node : nodes){
                nodeList.add(node);
            }
            setFatherHandler(nodeList);
        }

分別使用 fatherHandler ，sonHandler 這兩個 List 容器來保存任務編排程序中每一個任務節點的依賴父節點，系結下游子節點的資訊，可以采用如下的組織方式來維護一組任務節點流程，

// 任務節點間互相關聯代碼示例，
    node1.setSonHandler(node2,node3);
    node2.setSonHandler(node5).setFatherHandler(node1);
    node3.setSonHandler(node4).setFatherHandler(node1);
    node4.setSonHandler(node5).setFatherHandler(node3);
    node5.setFatherHandler(node2,node4);

圖7

上述代碼的組裝就實作了在記憶體中存盤圖片中對任務節點的依賴關系，完成了一次任務間作業流程關系的編排；有了任務的編排關系，就可以使用執行緒將整個任務流行起來，我們正式開始進入任務流程處理的環節，

圖8

任務流程處理器是每一個業務節點的流程運行，流程間控制的心臟，其運行時通用屬性與功能抽象如上圖所示，每一個流程在執行時都要關注處理自身的節點屬性隨流程的推進行狀態的變更，而這一個環節是每一個節點運行必須進行的，所以我們可以把這個重復性，統一的流程環節抽出來形成一個通用的流程處理器，后面的節點要想擁有該流程處理能力，只繼要繼承該類即可，

該類的定義結構整體如下：

@Data
    public abstract class SchedulingNode<T,V> {
          /**
         * 節點的名字 默認使用類名
         */
        protected String taskName = this.getClass().getSimpleName();
          /**
         * 節點執行結果狀態
         */
        ResultState status = ResultState.DEFAULT;
          /**
         * 將來要處理的任務節點的param引數
         */
        protected T param;    /**
         * 節點狀態
         */
        private static final int FINISH = 1;
        private static final int ERROR = 2;
        private static final int WORKING = 3;
        private static final int INIT = 0;
        /**
         * 節點狀態標記位
         */
        private final AtomicInteger state = new AtomicInteger(0);
          /**
         * 默認節點執行結果
         */
        private volatile WorkResult<V> workResult = WorkResult.defaultResult(getTaskName());
      
        /**
         * 任務執行方法
         * @param executorService
         * @param fromNode  來源于哪個父節點
         * @param remainTime  執行該節點剩余的時間
         * @param allParamUseNodes  全域所有的節點
         * @param taskContext   背景關系
         */
        public void execute(ExecutorService executorService, SchedulingNode fromNode, long remainTime,
                            ConcurrentHashMap<String, SchedulingNode<T,V>> allParamUseNodes,
                            TaskContext taskContext) {
          // 重復可復用的具體流程控制邏輯代碼
          ...
        }
    }

任務流程處理器 SchedulingNode 這個抽象類中做的重要的事通俗來說就是遍歷任務編排流中的各個節點，然后提交給線索程中去執行就可以，程序中需要控制節點的流轉的順序以及一些節點狀態保存，結果的保存，上述代碼中的 execute 方法就是該環節的入口，所有的結果存入在 allParamUseNodes map容器中，任務節點的狀態都是自身節點的屬性控制，整體環節的流程圖分享如下：

圖9

從流程圖中可以看到，整體是一條順序執行的流程，這里任務編排框架最主要的目標是要把每一個任務流中的節點執行完，為了實作任務間按編排好的順序與依賴的方向執行，增加了很多環節的控制，控制的方向采用逆向處理的手段，優先假設自己節點前有依賴，節點后有下游，需要雙方面關注他們的狀態，只有自身達到可執行的時機才會執行自己的功能，否則都是優先處理子節點，而實際呼叫子節點的方法最終也會執行這個 execute 方法入口，只是傳入的節點引數已做了變更，這一點類似不斷遍歷迭代的思想，來完成整個流程的運行，

現在回到最初提到的思考的問題3：任務流程始何跑起來，如何起到控制作用？實際從上述的流程圖中也能看到答案，就是本案例中采用到判斷節點的執行狀態來判斷的，因為這是跟實際業務需要有關的，任務編排了就是需要任務執行的程序有先有后，這樣才能滿足后一個任務依賴前一個任務的執行才可以的要求，不然大家都是并行執行的，所以任務本身會有一個非常重要的任務狀態，任務執行結果狀態這2個屬性，通過流的推進不斷判斷這些狀態是否達到滿足下個節點執行的前提條件從而起到了控制的作用，

這一個塊的流程推進，狀態判斷的偽代碼如下：

/*如果前方有依賴，存在兩種情況,以圖7中的編排任務流舉例說明
              一種是前面只有一個依賴，即 node2  ->  node5
             一種是前面有多個依賴，node2 node4 ->   node5，需要node2、node4都完成了才能輪到node5，
            */
            //只有一個依賴
            if (fatherHandler.size() == 1) {
                doDependsOneJob(fromNode, taskContext);
                runSonHandler(taskContext,executorService, remainTime, now);
            } else {
                //有多個依賴時
                doDependsJobs(executorService, fromNode, fatherHandler, now, remainTime, taskContext);
            }
          
       // 任務執行前，先做狀態的判斷，以起到控制的作用，
       private void doDependsOneJob(SchedulingNode dependWrapper, TaskContext taskContext) {
            if (ResultState.TIMEOUT == dependWrapper.getWorkResult().getResultState()) {
                workResult = timeoutResult();
                fastFail(INIT, null);
            } else if (ResultState.EXCEPTION == dependWrapper.getWorkResult().getResultState()) {
                workResult = defaultExResult(dependWrapper.getWorkResult().getEx());
                fastFail(INIT, null);
            } else {
                //前面任務正常完畢了，該自己了
                runSelf(taskContext);
            }
        }

業務節點是最終的任務執行，用于執行具體的業務耗時代碼，所有的業務邏輯都在這里類別中實作，一個編排的任務流中有多少個節點，那么就會有多少個業務節點實體，

業務節點為了復用業務執行器的控制邏輯，所以業務節點都會繼承任務流程處理器，業務節點需要宣告一些回呼函式，用于任務執行程序中的回呼鉤子，比如宣告出業務任務的核心法，用于任務流程處理器呼叫執行真正的業務內容，又比如當任務執行成功，或是失敗時是否需要回呼告知給業務方，自行用于狀態的感知與告警處理等，

業務節點的偽代碼如下：

@Component
    public class Node1 extends SchedulingNode<String,String> {
       
       @Resource
        private SampleService sampleService;
         
       @Override
        public Object task(TaskContext taskContext, String param) {
          // 具體的業務代碼邏輯實作
           // eg: db操作，http操作等
           sampleService.findItemById(...);
           ...
        }
      
        @Override
        public void onSuccess(TaskSupport support) {
        // 節點任務執行成功時的回呼
           ...
        }
    
        @Override
        public void onFail(TaskSupport support) {
          // 節點任務執行失敗時的回呼
       ...
        }
    }

這樣我們基本完成了這個任務編排框架的設計，最后可以如下來進行示例圖中的任務編排以及執行：

// 任務節點間互相關聯代碼示例，
    node1.setSonHandler(node2,node3);
    node2.setSonHandler(node5).setFatherHandler(node1);
    node3.setSonHandler(node4).setFatherHandler(node1);
    node4.setSonHandler(node5).setFatherHandler(node3);
    node5.setFatherHandler(node2,node4);
    
    // 設定背景關系
    TaskContext taskContext = new TaskContext();
    
    Map<String, SchedulingNode> results = TaskScheduling.start(5000L, taskContext, node1);
    ...

任務編排框架是一種用于管理和協調多個任務執行的技術，一些業務復雜的場景需要同時處理多個任務，這些任務之間可能有先后順序和依賴關系，此時通過任務編排技術可以很方便地組合這些任務，滿足業務需求，使用任務編排可以使得業務開發人員更加專注于業務本身，無需關注任務執行的細節，提高開發效率，改善編程體驗以及增強業務系統的可維護性，可擴展性，

本文分別從引入任務編排的場景以及如何設計一個任務編排框架兩個方面進行了闡述，核心是想與大家分享實作任務編排這一個工具類的框架需要思考的點，以及該框架是如何起到在任務并發執行的情況下進行流程控制的作用，并以流程圖的方式向大家展示了編排框架最核心的任務流轉控制程序，

本文在探究任務編排框架實作原理程序中，也參考了目前開源專案的實作，主要是 Gobrs-Async， async_QY-master ，在此向該開源專案及開源作者表示感謝，謝謝他們的無私分享和貢獻，同時我也想向大家分享這些專案，閱讀專案原始碼，學習專案的設計思想并運用到實際的專案中來 ; 優秀的開源專案鏈接見文章末尾的參考串列，

在文章的最后，我也思考了本文中描述的框架可以有以下的優化點：

• 任務編排的程序可以由圖形界面來承載，采用業務通用的 DGA 有向無環圖的資料結構來表述任務間的依賴關系
• 可以對任務執行緒池進行監控，來了解系統的負載，資源分配情況
• 可以引入動態執行緒程技術，不同的任務編排技術可以進行不同的配置，
• 可以有統一的日志追蹤組件，記錄任務整個生命周期的 debug 日志，方便排查任務例外的原因及問題定位，
• 可以將編排框架抽象成一個任務編排平臺，來承接更多的業務線的業務場景，減少重復造輪子，
• 可以將這個編排框架模塊載入到現有的定時任務調度平臺，將任務編排整合到任務調度平臺中，針對有業務關聯的多個任務進行依賴整合，使用任務的調度順序，時間更合理，同時也減少任務調度平臺的任務個數，減輕任務調度壓力， 減少任務間的互斥性等待等任務管理相關的復雜度設計，
• 業務關聯的任務編排成任務流，可以通過圖形的方式展示出業務間任務組合全貌，任務間可以輕松進行前后流程的修改，能方便后續業務的調整，應對業務的變更具有更強的擴展性，