→ popcorny: Task Dependency是DAG.. 可以直接找DAG的Library.. 11/17 15:07
→ swpoker: 哈~當作作業練習練習 XD 11/18 09:22
這不是流程的問題
是資料結構的問題
如果是自己寫程式的話
那就用popcorny的CompletableFuture來寫程式安排,比較有彈性
但是你想要用資料庫來驅動的話
那就會變成是tree結構了
因此要用爬節點的方式才能符合你的需求
所以我用爬節點的方式,還是使用ForkJoinPool來實作
就可以透過節點的關係來驅動實際的行為
---
其實爬節點可以用RecursiveTask實作,但有誰可以教學一下嗎?
我對於fork join還是不太會用
---
package com.swpoker;
import java.util.*;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
/**
*/
public class Test5 {
static class ExecuteBean{
public String name;
public String memo;
public int waittime;
public ExecuteBean(String name, String memo, int waittime ){
this.name=name;
this.memo=memo;
this.waittime=waittime;
}
}
static List<ExecuteBean> simulateExecute(){
List<ExecuteBean> rs = new ArrayList();
rs.add(new ExecuteBean("A","A1",1));
rs.add(new ExecuteBean("B","B1",12));
rs.add(new ExecuteBean("C","C1",2));
rs.add(new ExecuteBean("D","D1",4));
rs.add(new ExecuteBean("E","E1",5));
rs.add(new ExecuteBean("F","F1",2));
rs.add(new ExecuteBean("H","H1",10));
return rs;
}
static class DependBean{
public String name;
public String depend;
public DependBean(String name, String depend){
this.name=name;
this. depend=depend;
}
}
static List<DependBean> simulatDepend(){
List<DependBean> rs = new ArrayList();
rs.add(new DependBean("A",""));
rs.add(new DependBean("B","A"));
rs.add(new DependBean("C","A"));
rs.add(new DependBean("D","C"));
rs.add(new DependBean("E","B"));
rs.add(new DependBean("E","D"));
rs.add(new DependBean("F","C"));
rs.add(new DependBean("H","B"));
rs.add(new DependBean("H","C"));
return rs;
}
static List<ExecuteBean> findChildNode(String name , List<DependBean>
dependlist,List<ExecuteBean> executelist){
List<ExecuteBean> rs = new ArrayList();
dependlist.stream().filter( db -> db.depend.equals(name) ).forEach(db
-> rs.add(
executelist.stream().filter(e ->
e.name.equals(db.name)).findFirst().get()
));
return rs;
}
static class ExecuteActor {
ExecuteBean bean;
public ExecuteActor(ExecuteBean bean){
this.bean=bean;
}
public void dosomething(){
try{
long waittime =bean.waittime * 1000;
System.out.printf("%s start wait(%s)
%s\n",this.bean.memo,waittime, Calendar.getInstance().getTime());
//do something
Thread.sleep(waittime);
System.out.printf("%s end %s\n",this.bean.memo,
Calendar.getInstance().getTime());
}catch (Exception e){e.printStackTrace();}
}
}
private static Map<String, List<String>>
buildInVokeDepends(List<ExecuteBean> executelist, List<DependBean>
dependlist) {
Map<String, List<String>> dependrs = new HashMap();
executelist.stream().forEach( eb -> {
dependrs.put(eb.name,new ArrayList());
dependlist.stream().filter( de -> eb.name.equals(de.name) &&
Optional.ofNullable(de.depend).orElse("").isEmpty() == false ).forEach(
de -> dependrs.get(eb.name).add(de.depend)
);
} );
return dependrs;
}
private static ExecuteBean findFistExecute(List<DependBean> dependlist,
List<ExecuteBean> executelist) {
DependBean fisrtdb = dependlist.stream().filter(d
->Optional.ofNullable(d.depend).orElse("").isEmpty()).findFirst().get();
return
executelist.stream().filter(eb->eb.name.equals(fisrtdb.name)).findFirst().get();
}
static boolean invokeAndCheckExecuteBean(ExecuteBean
parentbean,ExecuteBean bean,Map<String,List<String>> invokedenpends){
if(parentbean == null){
return true;
}
synchronized (invokedenpends){
List<String> invokedenpend=invokedenpends.get(bean.name);
invokedenpend.remove(parentbean.name);
return invokedenpend.isEmpty();
}
}
static class NodeExecuteActor implements Runnable{
ExecuteBean parentbean ;
ExecuteBean bean;
ForkJoinPool forkJoinPool;
List<DependBean> dependlist;
List<ExecuteBean> executelist ;
Map<String, List<String>> inVokeDepends;
public NodeExecuteActor(ExecuteBean parentbean ,ExecuteBean
bean,ForkJoinPool forkJoinPool, List<DependBean> dependlist,List<ExecuteBean>
executelist,Map<String, List<String>> inVokeDepends){
this.parentbean=parentbean;
this.bean=bean;
this.forkJoinPool=forkJoinPool;
this.dependlist=dependlist;
this.executelist=executelist;
this.inVokeDepends=inVokeDepends;
}
@Override
public void run() {
System.out.printf("NodeExecuteActor %s start
%s\n",this.bean.name, Calendar.getInstance().getTime());
//check
if(invokeAndCheckExecuteBean(this.parentbean,this.bean,
this.inVokeDepends) == false){
return;
}
//excute self
new ExecuteActor( this.bean).dosomething();
//then invoke child
invokeChild(this.bean,this.forkJoinPool,
this.dependlist,this.executelist,this.inVokeDepends);
}
}
static void dealNode(ExecuteBean parentbean , ExecuteBean bean
,ForkJoinPool forkJoinPool, List<DependBean> dependlist,List<ExecuteBean>
executelist,Map<String, List<String>> inVokeDepends){
forkJoinPool.execute(new
NodeExecuteActor(parentbean,bean,forkJoinPool,dependlist,executelist,inVokeDepends));
}
static void invokeChild(ExecuteBean bean,ForkJoinPool forkJoinPool,
List<DependBean> dependlist,List<ExecuteBean> executelist,Map<String,
List<String>> inVokeDepends){
for(ExecuteBean eb : findChildNode(
bean.name,dependlist,executelist)){
dealNode(bean,eb,forkJoinPool,dependlist,executelist,
inVokeDepends);
}
}
static void startFirstNode(ExecuteBean bean ,ForkJoinPool forkJoinPool,
List<DependBean> dependlist,List<ExecuteBean> executelist,Map<String,
List<String>> inVokeDepends){
// invokeChild(bean,forkJoinPool,
dependlist,executelist,inVokeDepends);
dealNode(null, bean , forkJoinPool, dependlist,executelist,
inVokeDepends);
}
public static void main(String [] args) throws Exception{
System.out.printf("main start %s\n",
Calendar.getInstance().getTime());
List<ExecuteBean> executelist = simulateExecute();
List<DependBean> dependlist=simulatDepend();
Map<String,List<String>>
invokedepends=buildInVokeDepends(executelist, dependlist);
ExecuteBean firstBean = findFistExecute(dependlist,executelist);
ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
//start
startFirstNode(firstBean,forkJoinPool,dependlist,executelist,invokedepends);
forkJoinPool.shutdown();
forkJoinPool.awaitTermination(1, TimeUnit.DAYS);
System.out.printf("main end %s\n", Calendar.getInstance().getTime());
}
}
※ 引述《PTTCATKING (懷念美國貓王)》之銘言：
: 現在一個流程是這樣
: Thread A 執行完成之後，分成兩條Thread，同時執行 B跟 C
: 然後C這條線執行完後，會再執行D
: 等到B這條線跟 C&D這條線兩條都跑完後，才開始執行E
: 如果像是我以下這種寫法，算是 B跟 CD跑完之後，才開始跑E嗎
: 那一開始起頭的 A 我要怎麼寫呢
: public void execute(TaskExecutionContext executor) throws RuntimeException {
: System.out.println("Thread E ");
: Thread threadB = new Thread(new Runnable() {
: public void run() {
: try {
: System.out.println("Thread B 開始..");
: for(int i = 0; i < 5; i++) {
: Thread.sleep(1000);
: System.out.println("Thread B 執行..");
: }
: System.out.println("Thread B 即將結束..");
: }
: catch(InterruptedException e) {
: e.printStackTrace();
: }
: }
: });
: Thread threadC = new Thread(new Runnable() {
: public void run() {
: try {
: System.out.println("Thread C 開始..");
: for(int i = 0; i < 5; i++) {
: Thread.sleep(1000);
: System.out.println("Thread C 執行..");
: }
: System.out.println("Thread C 即將結束..");
: System.out.println("Thread D 開始..");
: for(int i = 0; i < 5; i++) {
: Thread.sleep(1000);
: System.out.println("Thread D 執行..");
: }
: System.out.println("Thread D 即將結束..");
: }
: catch(InterruptedException e) {
: e.printStackTrace();
: }
: }
: });
: threadB.start();
: threadC.start();
: try {
: // Thread B 加入 Thread A
: threadB.join();
: threadC.join();
: }
: catch(InterruptedException e) {
: e.printStackTrace();
: }
: System.out.println("Thread E 執行");
: for (int i = 1; i <= 30; i++) {
: System.out.println("Task 1 says: " + i + executor);
: executor.setStatusMessage("第一個任務執行到第 " + i + " 個");
: executor.setCompleteness(i / 30D);
: try {
: Thread.sleep(1000);
: } catch (InterruptedException e) {
: ;
: }
: executor.pauseIfRequested();
: if (executor.isStopped()) {
: break;
: }
: }
: }
: 格式可能有點跑掉，這是 讓E等待 B跟CD的寫法
: 怎麼寫成從A開始呢
: 因為 E 就是主程序了說..
※ 引述《PTTCATKING (懷念美國貓王)》之銘言：
: 假設有二十個任務要跑
: 而他在DB裡會有兩組編號(欄位名稱)
: 第一組是1~20的流水號
: 第二組是執行優先順序(記錄需等待的對象)
: A 跑完 同時執行 B & CD 兩條線同時跑，而B會先跑完，但要等CD都跑完後，BCD都結束
: 才開始跑E
: A的編號就是 1 0 (無等待對象)
: B的編號就是 2 1 (等待A，1是A的編號)
: C的編號就是 3 1 (等待A，1是A的編號)
: D的編號就是 4 3 (等待C，3是C的編號)
: E的編號就是 5 2 (等待B，2是B的編號)
: E的編號就是 5 4 (等待D，4是D的編號)
: 因為E要等兩個，所以在DB裡 E 會有兩筆資料
: 請問 有沒有什麼現有套件有能夠執行這件事情呢
: 設定那兩個編號，則程式執行後就按照兩個編號，決定程式優先順序，並有可能同時執行
: 多程序
: 以後維護就是設定編號就能決定跑程式的優先順序
: 今天用CountDownLatch寫出類似的東西，被認為是寫死的東西orz
: 這是用排程器???或是什麼方向呢?? @O@
: 能否有高手指點明燈給我方向，謝謝 T_T
--
※ 發信站: 批踢踢實業坊(ptt.cc), 來自: 163.29.29.131
※ 文章網址: http://www.ptt.cc/bbs/java/M.1416196414.A.4B7.html
※ 編輯: swpoker (163.29.29.131), 11/17/2014 13:41:10