图节点（任务）的并行执行并找到关键任务

Question

我已经使用 Java 实现了一个有向图。它适用于项目规划者，一个节点代表具有不同属性的任务。我已经成功实现了拓扑排序，但我需要一种方法来在任务的依赖项完成后立即运行/执行并行任务。

这是我的实现：

import java.util.ArrayList;
import java.util.List;
import java.util.*;

public class Task implements Comparable<Task> {
    int number;
    String name;
    int time;
    int staff;
    int earliestStart, latestStart;
    List<Integer> dependencies;
    List<Task> outEdges;
    int cntPredecessors;
    Status status;
    public enum Status {UNVISITED,RUNNING,VISITED};
    @Override
    public String toString() {
        return "Task{" +
                "number=" + number +
                ", name='" + name + '\'' +
                ", time=" + time +
                ", staff=" + staff +
                ", dependencies=" + dependencies +
                '}';
    }

    public Task(int number, String name, int time, int staff) {
        setNumber(number);
        setName(name);
        setTime(time);
        setStaff(staff);
        dependencies=new ArrayList<>();
        outEdges=new ArrayList<>();
        status = Status.UNVISITED;
     }

    public int getNumber() {
        return number;
    }

    public void setNumber(int number) {
        this.number = number;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getTime() {
        return time;
    }

    public void setTime(int time) {
        this.time = time;
    }

    public int getStaff() {
        return staff;
    }

    public void setStaff(int staff) {
        this.staff = staff;
    }

    public List<Integer> getDependencies() {
        return dependencies;
    }

    public void setDependencies(List<Integer> dependencies) {
        this.dependencies = dependencies;
    }

    public List<Task> getOutEdges() {return outEdges; }

    public void setOutEdge(Task t) {outEdges.add(t); }



    public int getIndegrees() { return cntPredecessors; }

    public void setIndegree() { cntPredecessors = dependencies.size();}

    public Status getStatus() {return this.status; }

    public Task findTaskWithNoInDegree() {
            if (this.cntPredecessors == 0) return this;
            return null;
    }
    

    public int compareTo(Task other) {

        return Integer.compare(this.time, other.time);

        }
} //END class Task

// The class Main represents the Task objects in a graph

import java.util.*;

public class Main {
    static int maxnr = 0;

    public static void main(String[] args) {
        Map<Integer, Task> map=new HashMap<>();
        Scanner scanner = new Scanner(Main.class.getResourceAsStream("/res/house.txt"), "UTF-8");
        
        Main mainObject = new Main();
        map = mainObject.fromScanner(scanner);
        System.out.println("DEBUG: maxnr " + maxnr);
        mainObject.setInDegrees(map);
        mainObject.setOutEdges(map);

        //System.out.println("DEBUG: Size of outEdges for Task 1 is : " + map.get(1).getOutEdges().size());
        //System.out.println("DEBUG: Indegrees for Task 8 is : " + map.get(8).getIndegrees());
        mainObject.topSort(maxnr,map);

        for(Integer k:map.keySet()) {
            //System.out.println("DEBUG outEdges for Task number " + map.get(k).getNumber() + " " + map.get(k).getOutEdges());
        }

    } // END of void main(String[] args)


    public void setInDegrees(Map<Integer, Task> map) {
        for(Integer k:map.keySet()) {
            Task task = map.get(k);
            task.setIndegree();
        }
    }


    public void setOutEdges(Map<Integer, Task> map) {
        for(Integer k:map.keySet()) {
            // map.get(k).setIndegrees();
            for(Integer dep:map.get(k).getDependencies()) {
                //System.out.println("DEBUG: "+ dep);
                //System.out.print(" DEBUG:  Name is "  + map.get(dep).getName());
                map.get(dep).setOutEdge(map.get(k));
            }
            //System.out.println(map.get(k));
        }
    } //END of setOutEdges()
        // toplogical sort # Big O(|V| +|E|)  for indegree calc and since the code only looks at each edge once!
        // S is Set of all nodes with no incoming edges
    public void topSort(int maxnr, Map<Integer, Task> map) {
        ArrayList<Task> L = new ArrayList<Task>(maxnr);
        //LinkedList<Task> L = new LinkedList<>();
        //HashSet<Task> S = new HashSet<>(maxnr);
        LinkedList<Task> S = new LinkedList<>();

        for(Integer n:map.keySet()) {
            if(map.get(n).getIndegrees() == 0) {
                S.add(map.get(n));
            }
        }
        System.out.println("DEBUG: Set S is " + S);
        //HashSet<Task> S2 = new HashSet<>(S);

        Task t;
        int counter= 0;
        while(!S.isEmpty()) {
            //System.out.print("Topsort: Task and S. " + t.getNumber());
            t = S.iterator().next();
            S.remove(t);
            //System.out.print("Topsort : " + t.getNumber());
            L.add(t);
            //System.out.println("Starting " + t.getNumber());
            counter++;

            for(Iterator<Task> it = t.outEdges.iterator(); it.hasNext();) {
                Task w =  it.next();
                w.cntPredecessors--;
                if (w.getIndegrees() == 0) {
                    S.add(w);
                   // System.out.println("Starting " + w.getNumber());
                }
            }
        }
        System.out.println();

        if (counter < maxnr) {
            System.out.println("Cycle detected, topsort not possible");
        } else {
            //System.out.println("Topsort : " + Arrays.toString(L.toArray()));
            Iterator<Task> topsortIt = L.iterator();
            System.out.print("\n Topsort list is: ");
            while (topsortIt.hasNext()) {
                System.out.print(" " + topsortIt.next().getNumber());
            }
            System.out.println();
        }
    } //END of topSort()

    public Map fromScanner(Scanner scanner) {
    Map<Integer, Task> map=new HashMap<>();
    maxnr = scanner.nextInt();
    while (scanner.hasNextLine()) {
        String line=scanner.nextLine();
        if (line.isEmpty() ) continue;
        Scanner s2=new Scanner(line);
        Task task = new Task(s2.nextInt(), s2.next(), s2.nextInt(), s2.nextInt());
        while (s2.hasNextInt()) {
            int i = s2.nextInt();
            if (i != 0) {
                task.getDependencies().add(i);
            }
        }
        map.put(task.getNumber(), task);
    }
    return map;
    } //END of fromScanner()

    } //END of class Main

house.txt 的内容：第一行（数字）是最大节点/任务。这些列是： 任务编号、名称、完成所需时间、人力需求、依赖边缘（以 0 结尾）。

8
1   Build-walls     4 2       5       0
2   Build-roofs     6 4       1       0
3   Put-on-wallpapers   1 2       1       2       0
4   Put-on-tiles        1 3       2       0
5   Build-foundation    4 2       0
6   Make-floor          2 2       5       0
7   Put-carpet-floor    4 2       6       2       0
8   Move-in         4 4       3       7       0

应该首先启动没有传入边（即没有依赖项）的任务。 应该打印任务的执行，例如上面给出的输入：

Time: 0      Starting: 5   // Task 5 only one with no  dependencies
        Current staff: 2 

Time: 4      Finished: 5
             Starting: 6
             Starting: 1
        Current staff: 4   // sum of manpower from Task 6 and 1 => 2 +  2 = 4

Time: 6     Finished: 6
       Current staff: 2    

Time: 8     Finished: 1
            Finished: 1  
            Starting: 2
            Starting: 3 
       Current staff: 6

等等

Answer 1

如果它对任何人有帮助，这正是 JavaRed Library 的用例 - 在定义为图的流中调度和运行异步任务。

简而言之，一个像下面这样复杂的图流：

可以简单地实现：

Result<String> aResult = produceFutureOf(String.class).byExecuting(() -> executeA());
Result<String> bResult = ifResult(aResult).succeed().produceFutureOf(String.class).byExecuting(a -> executeB(a));
Result<String> cResult = ifResult(aResult).succeed().produceFutureOf(String.class).byExecuting(a -> executeC(a));
Result<String> eResult = ifResult(aResult).succeed().produceFutureOf(String.class).byExecuting(a -> executeE(a));
Result<String> dResult = ifResults(bResult, cResult).succeed().produceFutureOf(String.class).byExecuting((b, c) -> executeD(b, c));
Result<String> fResult = ifResult(eResult).succeed().produceFutureOf(String.class).byExecuting(e -> executeF(e));
Result<String> gResult = ifResult(fResult).succeed().produceFutureOf(String.class).byExecuting(f -> executeG(f));
return ifResults(dResult, gResult).succeed().produceFutureOf(String.class).byExecuting((d, g) -> executeH(d, g));

更多信息请关注project wiki

Answer 2

有几种方法。

您可以使用 Java 8 CompletableFuture 或 Guava ListenableFuture 来实现它。

您的任务应该实现Runnable 或Callable。当你执行当前任务时，你应该递归地执行它的preTasks（如果任务是复合模式，它会知道它是前任）并确保它们已经完成（通过标志来表达状态或其他东西）。

像这样：

CompletableFuture.allOf(predecessor).thenRunAsync(current)...
Futures.allAsList(predecessor)...

Answer 3

要安排您的任务，您可以自己维护一个线程池，也可以让一个库为您完成。如果你自己维护线程池，那就很简单了（原谅可怜的伪代码）：

assume you already have tasks topoligically sorted : t0,t1,t2,....,tn
int next_task=0;  //global pointer to next task

then in each thread, you do:
    while (true) {
        atomic {
            if (next_task > n) break;
            t = get_task (next_task);
            next_task = next_task + 1;
        }
        run task t;
    }

通过这种方式，您可以按照尊重其依赖关系的顺序执行所有任务，并且每个线程将在完成最后一个任务后立即跳转到下一个任务。

如果你想要一个库来为你安排它，也许是 OpenMP、TBB，或者查看这个线程 How to Implement a DAG-like Scheduler in Java?？

希望这会有所帮助。