本文主要介紹在之家廣告業務系統中運用任務編排治理工具的場景及其可以解決的問題，講解任務編排架構的核心要點，以及向大家示範一個任務編排架構的基本結構，讓大家對任務編排工具增強業務開發效率，提高研發品質有更深刻的了解。
1.背景
我們開始以下面的實際業務場景代入任務編排工具的使用。
廣告引擎流量接入層中一個使用者流量請求過來，系統會對流量進行一系列的驗證，資料加工處理最終會将使用者流量轉化為廣告引擎可召回的廣告請求，具體過程如下：
圖1
從圖 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 個環節都執行完成後，下面可以進行的結果彙總過程
}
1.1
存在的問題分析
●存在兩個多線程異步并發程式設計代碼相似的處理流程，如果業務場景再複雜一些，比如有 3 個或是 4 個并發場景呢？或是某一個并發場景中存在任務上下依賴的情況，可以想象在這個過程中，運作多線程程式設計的代碼也會變成特别臃腫，多線程處理那套模闆會出現多次，且無法清晰地看清任務的流程。程式設計人員針對這種複雜的業務流程要處理兩大類的事務，一類是多線程技術運用對任務流程的控制，一類是業務本身的邏輯實作。
●兩個多線程異步并發程式設計代碼段，比如 TaskA, TaskB 包含的子任務都實作了多線程異步并發程式設計。但 TaskA 與 TaskB 之間是無法并發程式設計的。需要 TaskA 那段完成之後，才能順序執行 TaskB 那段。也就是 TaskB 這一整塊需要等待 TaskA 整個任務的完成。但實際組裝整個商詳頁面資料，這兩個任務間是可以不用彼此等待的， 是以會産生等待時間影響系統整體 QPS。
是以遇到業務複雜的場景，我們會考慮采用分而治之的思想把業務拆解成一個個子任務塊來執行，并會使用多線程技術來提效實作整個業務流轉。此時多線程程式設計技術的運用也會變得複雜，在實際任務間存在依賴關系，并發限制，逾時和錯誤處理，多階段任務等情況需要一并解決的場景下，單純地采用傳統的多線程程式設計技術（CompletableFuture）會讓程式設計體驗下降很多，任務間，資料間的依賴并不清晰，後續維護時需要針對業務中的任務塊進行更換、順序調整或者由串行改并行時需要增加很多重構工作。
不同業務場景中的任務拆分之後，其組合的的場景也是豐富的，很多時候是串行&并行并存的場景，比如如下幾個例子：
圖3
是以基于上述痛點分析後，如果能有一種多線程程式設計技術封裝，能将多個任務按照依賴關系和執行順序組合起來的技術，這樣業務程式設計人員在實際業務實作中能更加專注于業務本身，無需關注多線程代碼結構的維護，任務執行的細節，而且還能提高開發效率和業務系統的可維護性，那将會對研發體驗與研發品質都将有質的提升。
而這就是任務編排技術的範疇。
2.任務編排介紹
任務編排是一種将多個任務按照依賴關系和執行順序組合起來的技術。可以把"任務"這個原子機關按照自己的方式進行編排，任務之間可能互相依賴，複雜一點的編排之後就能形成一個工作流。
在業務系統開發中，一些業務複雜的場景需要同時處理多個任務，這些任務之間可能有先後順序和依賴關系，此時通過任務編排技術可以很友善地組合這些任務，滿足業務需求。
與傳統多線程程式設計相比，任務編排的最大優點是易于排程和控制，因為任務編排架構通常會提供一組方法接口來處理任務之間的依賴關系和執行順序，程式設計人員隻需要根據任務的性質和優先級來描述整個任務流程，簡化了管理和排程的複雜性。
相反，對于隻利用多線程去處理任務的方式，程式擁有複雜的線程控制邏輯，修改各個線程共享資料的方法，有大量重複的多線程程式設計代碼，這将增加程式設計難度和程式的不穩定性。
此外，使用任務編排的模型，可以根據任務流程進行系統分析，提出資源高峰并進行優化，進而提高系統效率。
總之，這些業務開發場景都需要用到任務編排，使用任務編排可以使得業務開發人員更加專注于業務本身，無需關注任務執行的細節，提高開發效率，改善程式設計體驗以及增強業務系統的可維護性。
3.任務編排架構核心探究
在遇到業務複雜的場景需要同時處理多個任務的場景下，我們需要優先考慮實作一個任務編排架構來支撐原子任務按業務流來進行任務流轉，優化線程之間的互動，確定任務按正确的順序執行，進而避免進入複雜多線程程式設計的泥潭。那我們現在可以思考如何實作一個支援任務編排工作流的架構，這個事項中會有哪些思考呢？
以下是我列舉要實作一個任務編排架構需要思考的問題：
1. 任務間的依賴關系如何維護，用什麼方式可以存儲這些原子任務間的彼此關系？
2. 任務間的資訊如何傳遞，任務參數如何透傳，任務結果如何傳遞給下一個任務？
3. 任務流程始何跑起來，如何起到控制作用；先執行什麼，後執行什麼，如何實作任務間并行執行，又是如何控制并發執行任務的前後順序的？
4. 任務運作中出現了異常怎麼感覺，怎麼處理，會對整個流程有什麼影響？
下面我們針對這4類問問整體進行一個個拆解，并配以流程圖及示例代碼來講解具體的架構設計，以及簡潔明了的解釋為何這樣設計;在開始介紹前我們先有以下的前提
●前提一：該任務編排架構定位于業務系統開發過程中，用于替代傳統多線程程式設計流程控制那個環節，目标是應用架構工具後，業務開發人員更加專注于業務本身，無需關注任務并行執行，任務流程控制的細節。這些場景有别于我們熟知且常見的大資料分析平台中的任務編排場景。
●前提二：我們已經對業務系統的原子性任務己經有了完善合理的拆分，并成功的梳理出任務間的上下依賴關系。
●前提三：這将是一個小而精的輕量級任務編排，任務并發執行的流程控制架構，你可以了解其就是對常用 JUC 工具包的二次包裝，是可以提取成一種工具類，也可以抽象成一種任務編排并發執行平台。後續内容将向大家講解流程編排這個環節的核心設計。
3.1
整體架構
圖4
整個架構需要的角色相對較少，主要包含任務啟動器，任務流程解析器，任務流程處理器，任務總線（TaskContext）以及業務任務塊組成。
任務啟動器扮演入口角色，負責整個流程的啟動，是與其它系統子產品互動的門戶。
任務流程解析器記錄着所有業務任務塊間的前後依賴關系，可以了解是一堆配置，這部配置設定置資訊可以由多種資料容器來承載。
任務流程處理器是這個任務編排的心髒，其負責從任務流程解析器中解讀出編排好的任務間關系，并統一采用多線程并發程式設計技術來并發執行這些任務，需要對任務進行前後執行順序進行控制，并需要對在運作中任務出現了異常進行感覺。
業務任務塊中是真正業務功能實作者，其聲明多個回調方法讓任務流程處理器在适當的時機進行調用，進而來完成業務需求的功能，最終達到業務需求目錄。
任務總線會貫穿整體任務編排并發執行過程的始終，用于傳遞任務啟動時傳入的任務參數，任務配置等資訊。
當所有任務執行完畢後，所有任務的執行結果會彙總，用于最終的業務傳回。
3.2
任務啟動入口
任務編排的啟動者，啟動類 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;
}
3.3
任務流程解析器
任務流程解析器子產品是整體任務編排架構重要的的子產品，主要有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
上述代碼的組裝就實作了在記憶體中存儲圖檔中對任務節點的依賴關系，完成了一次任務間工作流程關系的編排；有了任務的編排關系，就可以使用線程将整個任務流行起來，我們正式開始進入任務流程處理的環節。
3.4
任務流程處理器
圖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);
}
}
3.5
業務節點
業務節點是最終的任務執行，用于執行具體的業務耗時代碼，所有的業務邏輯都在這裡類别中實作，一個編排的任務流中有多少個節點，那麼就會有多少個業務節點執行個體。
業務節點為了複用業務執行器的控制邏輯，是以業務節點都會繼承任務流程處理器，業務節點需要聲明一些回調函數，用于任務執行過程中的回調鈎子，比如聲明出業務任務的核心法，用于任務流程處理器調用執行真正的業務内容，又比如當任務執行成功，或是失敗時是否需要回調告知給業務方，自行用于狀态的感覺與告警處理等。
業務節點的僞代碼如下：
@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);
...
4.總結
任務編排架構是一種用于管理和協調多個任務執行的技術，一些業務複雜的場景需要同時處理多個任務，這些任務之間可能有先後順序和依賴關系，此時通過任務編排技術可以很友善地組合這些任務，滿足業務需求。使用任務編排可以使得業務開發人員更加專注于業務本身，無需關注任務執行的細節，提高開發效率，改善程式設計體驗以及增強業務系統的可維護性，可擴充性。
本文分别從引入任務編排的場景以及如何設計一個任務編排架構兩個方面進行了闡述，核心是想與大家分享實作任務編排這一個工具類的架構需要思考的點，以及該架構是如何起到在任務并發執行的情況下進行流程控制的作用，并以流程圖的方式向大家展示了編排架構最核心的任務流轉控制過程。
本文在探究任務編排架構實作原理過程中，也參考了目前開源項目的實作，主要是 Gobrs-Async， async_QY-master ，在此向該開源項目及開源作者表示感謝，謝謝他們的無私分享和貢獻，同時我也想向大家分享這些項目，閱讀項目源碼，學習項目的設計思想并運用到實際的項目中來 ; 優秀的開源項目連結見文章末尾的參考清單。
在文章的最後，我也思考了本文中描述的架構可以有以下的優化點：
- 任務編排的過程可以由圖形界面來承載，采用業務通用的 DGA 有向無環圖的資料結構來表述任務間的依賴關系
- 可以對任務線程池進行監控，來了解系統的負載，資源配置設定情況
- 可以引入動态線程程技術，不同的任務編排技術可以進行不同的配置。
- 可以有統一的日志追蹤元件，記錄任務整個生命周期的 debug 日志，友善排查任務異常的原因及問題定位。
- 可以将編排架構抽象成一個任務編排平台，來承接更多的業務線的業務場景，減少重複造輪子。
- 可以将這個編排架構子產品載入到現有的定時任務排程平台，将任務編排整合到任務排程平台中，針對有業務關聯的多個任務進行依賴整合，使用任務的排程順序，時間更合理，同時也減少任務排程平台的任務個數，減輕任務排程壓力， 減少任務間的互斥性等待等任務管理相關的複雜度設計。
- 業務關聯的任務編排成任務流，可以通過圖形的方式展示出業務間任務組合全貌，任務間可以輕松進行前後流程的修改。能友善後續業務的調整，應對業務的變更具有更強的擴充性。
參考
開源項目：async_QY-master
開源項目：asyncTool
開源項目：Gobrs-Async
開源項目：powerjob
作者簡介
劉元軍
■ 主機廠事業部-技術部-廣告技術團隊。
■ 2017年加入汽車之家, 目前主要負責DSP廣告及主機廠線索業務後端開發工作，熱衷于WEB服務領域新技術的探索與實踐。
來源:微信公衆号:之家技術
出處:https://mp.weixin.qq.com/s/ZhznXZ1NXPFwXdUnAkidSA