本文主要介紹在之家廣告業務系統中運用任務編排治理工具的場景及其可以解決的問題，講解任務編排框架的核心要點，以及向大家演示一個任務編排框架的基本結構，讓大家對任務編排工具增強業務開發效率，提高研發質量有更深刻的理解。 ...
本文主要介紹在之家廣告業務系統中運用任務編排治理工具的場景及其可以解決的問題，講解任務編排框架的核心要點，以及向大家演示一個任務編排框架的基本結構，讓大家對任務編排工具增強業務開發效率，提高研發質量有更深刻的理解。
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
作者|劉元軍
本文來自博客園，作者：古道輕風，轉載請註明原文鏈接：https://www.cnblogs.com/88223100/p/Exploring-the-Application-of-Task-Orchestration-Tools-in-Business-Systems.html
