這篇文章主要講解了“DAG實(shí)現(xiàn)任務(wù)調(diào)度以及優(yōu)化”，文中的講解內(nèi)容簡(jiǎn)單清晰，易于學(xué)習(xí)與理解，下面請(qǐng)大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“DAG實(shí)現(xiàn)任務(wù)調(diào)度以及優(yōu)化”吧！
專注于為中小企業(yè)提供成都網(wǎng)站設(shè)計(jì)、成都做網(wǎng)站、外貿(mào)網(wǎng)站建設(shè) 服務(wù),電腦端+手機(jī)端+微信端的三站合一,更高效的管理,為中小企業(yè)開平免費(fèi)做網(wǎng)站 提供優(yōu)質(zhì)的服務(wù)。我們立足成都，凝聚了一批互聯(lián)網(wǎng)行業(yè)人才，有力地推動(dòng)了近1000家企業(yè)的穩(wěn)健成長(zhǎng)，幫助中小企業(yè)通過網(wǎng)站建設(shè)實(shí)現(xiàn)規(guī)模擴(kuò)充和轉(zhuǎn)變。
DAG Task 任務(wù)調(diào)度算法 實(shí)現(xiàn) github:https://github.com/smartxing/algorithm
1 有向圖的構(gòu)建
DAG dag = new DAG();
dag.addVertex("A");
dag.addVertex("B");
dag.addVertex("C");
dag.addVertex("D");
dag.addEdge("A", "B");
dag.addEdge("A", "C");
System.out.println(dag); 2 拓?fù)渑判驒z測(cè)圖中是否有環(huán)
public boolean isCircularity() {
Set set = inDegree.keySet();
//入度表
Map inDegree = set.stream().collect(Collectors
.toMap(k -> k, k -> new AtomicInteger(this.inDegree.get(k).size())));
//入度為0的節(jié)點(diǎn)
Set sources = getSources();
LinkedList queue = new LinkedList();
queue.addAll(sources);
while (!queue.isEmpty()) {
Object o = queue.removeFirst();
outDegree.get(o)
.forEach(so -> {
if (inDegree.get(so).decrementAndGet() == 0) {
queue.add(so);
}
});
}
return inDegree.values().stream().filter(x -> x.intValue() > 0).count() > 0;
} 3 stage優(yōu)化
eg
如果任務(wù)存在如下的關(guān)系 ， task1 執(zhí)行完后執(zhí)行 task2 ,task2 執(zhí)行完后執(zhí)行task3 ...
Task1 -> Task2 -> Task3 -> Task4
這些task 本來就要串行執(zhí)行的 可以把這些task 打包在一塊 減少線程上下文的切換
eg : 復(fù)雜一點(diǎn)的DAG:
/**
* H
* \
* G
* \
* A -> B
* \
* C- D -E - F-> J
*
*
*
* 優(yōu)化后得 ==>
*
* (H,G)
* \
* A -> B
* \
* (C,D,E) - (F,J)
*
*/
詳見chain方法： 關(guān)鍵代碼如下
private void chain_(Set sources, final LinkedHashSetMultimap foutChain, final LinkedHashSetMultimap finChain) {
sources.forEach(sourceNode -> {
ArrayList maxStage = Lists.newArrayList();
findMaxStage(sourceNode, maxStage);
if (maxStage.size() > 1) { //存在需要合并的stage
addVertex(foutChain, finChain, maxStage);//添加一個(gè)新節(jié)點(diǎn)
Object o = maxStage.get(maxStage.size() - 1); //最后一個(gè)節(jié)點(diǎn)
reChain_(foutChain, finChain, maxStage, o);
}
if (maxStage.size() == 1) {
//不存在需要合并的stage
addVertex(foutChain, finChain, sourceNode);//添加一個(gè)新節(jié)點(diǎn)
Set subNodes = outDegree.get(sourceNode);
addSubNodeage(foutChain, finChain, sourceNode, subNodes);
}
});
}
4 測(cè)試DAG 執(zhí)行
測(cè)試程序: 詳見 DAGExecTest
1 新建一個(gè)task 只打印一句話
public static class Task implements Runnable {
private String taskName;
public Task(final String taskName) {
this.taskName = taskName;
}
@Override public void run() {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("i am running my name is " + taskName + " finish ThreadID: " + Thread.currentThread().getId());
}
public String getTaskName() {
return taskName;
}
@Override public String toString() {
return taskName;
}
}
2 構(gòu)建DAG
DAG dag = DAG.create();
Task a = new Task("a");
Task b = new Task("b");
Task c = new Task("c");
Task d = new Task("d");
Task e = new Task("e");
Task f = new Task("f");
Task g = new Task("g");
Task h = new Task("h");
Task j = new Task("j");
dag.addVertex(a);
dag.addVertex(b);
dag.addVertex(c);
dag.addVertex(d);
dag.addVertex(e);
dag.addVertex(f);
dag.addVertex(g);
dag.addVertex(h);
dag.addVertex(j);
dag.addEdge(h, g);
dag.addEdge(g, b);
dag.addEdge(a, b);
dag.addEdge(b, f);
dag.addEdge(c, d);
dag.addEdge(d, e);
dag.addEdge(e, f);
dag.addEdge(f, j);
構(gòu)建完成后如圖
* H
* \
* G
* \
* A -> B
* \
* C- D -E - F-> J
3 stage 切分
DAG chain = dag.chain();
執(zhí)行完圖入下：
* (H,G)
* \
* A -> B
* \
* (C,D,E) - (F,J)
4 執(zhí)行 DAG DAGExecTest 最終結(jié)果打印如下如下：
可以發(fā)現(xiàn)有3個(gè)Stage stage1 包含3個(gè)task task分別在不同的線程里面執(zhí)行
其中c-d-e g-c f-j是經(jīng)過優(yōu)化的在同一個(gè)線程里面執(zhí)行，減少了不必要的上下文切換
i am running my name is a finish ThreadID: 10
i am running my name is c finish ThreadID: 11
i am running my name is h finish ThreadID: 12
i am running my name is d finish ThreadID: 11
i am running my name is g finish ThreadID: 12
i am running my name is e finish ThreadID: 11
stage 結(jié)束 ： task detached:a, task chain c-d-e task chain h-g
-----------------------------------------------
i am running my name is b finish ThreadID: 14
stage 結(jié)束 ： task detached:b,
-----------------------------------------------
i am running my name is f finish ThreadID: 11
i am running my name is j finish ThreadID: 11
stage 結(jié)束 ： task chain f-j
測(cè)試執(zhí)行關(guān)鍵代碼如下：
chain.execute(col -> {
Set set = (Set) col;
List completableFutures = Lists.newArrayList();
StringBuilder sb = new StringBuilder();
set.stream().forEach(x -> {
if (x instanceof Task) {
CompletableFuture future = CompletableFuture.runAsync((Task) x, executorService);
completableFutures.add(future);
sb.append(" task detached:" + ((Task) x).getTaskName()).append(",");
}
if (x instanceof List) {
List taskList = (List) x;
CompletableFuture future = CompletableFuture.runAsync(()->
taskList.forEach(Task::run));
completableFutures.add(future);
sb.append(
" task chain " + Joiner.on("-").join(taskList.stream().map(Task::getTaskName).collect(Collectors.toList())));
}
});
CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[completableFutures.size()])).join();
System.out.println("stage 結(jié)束 ： " + sb.toString());
System.out.println("-----------------------------------------------");
}); 感謝各位的閱讀，以上就是“DAG實(shí)現(xiàn)任務(wù)調(diào)度以及優(yōu)化”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對(duì)DAG實(shí)現(xiàn)任務(wù)調(diào)度以及優(yōu)化這一問題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是創(chuàng)新互聯(lián)，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！
分享文章：DAG實(shí)現(xiàn)任務(wù)調(diào)度以及優(yōu)化
當(dāng)前網(wǎng)址：
http://weahome.cn/article/jssdec.html
重慶分公司
重慶分公司
