Java: 循环等待直到ThreadPoolExecutor的任务完成再继续
Java: Wait in a loop until tasks of ThreadPoolExecutor are done before continuing
我正在努力使 Dijkstra 算法并行化。使每个节点线程查看当前节点的所有边缘。这与线程并行，但开销太大。这导致比算法的顺序版本更长的时间。
添加了 ThreadPool 来解决这个问题，但我无法等待任务完成才能继续下一次迭代。只有完成一个节点的所有任务后，我们才能继续。在我可以按节点搜索下一个最近的任务之前，我们需要所有任务的结果。
我试过 executor.shutdown() 但使用这种方法它不会接受新任务。我们如何才能在循环中等待直到每个任务完成，而不必每次都声明 ThreadPoolExecutor。这样做会破坏通过使用它而不是常规线程来减少开销的目的。
我想到的一件事是添加任务（边缘）的 BlockingQueue。但是对于这个解决方案，我一直在等待任务在没有 shudown() 的情况下完成。
public void apply(int numberOfThreads) {
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(numberOfThreads);
class DijkstraTask implements Runnable {
private String name;
public DijkstraTask(String name) {
this.name = name;
}
public String getName() {
return name;
}
@Override
public void run() {
calculateShortestDistances(numberOfThreads);
}
}
// Visit every node, in order of stored distance
for (int i = 0; i < this.nodes.length; i++) {
//Add task for each node
for (int t = 0; t < numberOfThreads; t++) {
executor.execute(new DijkstraTask("Task " + t));
}
//Wait until finished?
while (executor.getActiveCount() > 0) {
System.out.println("Active count: " + executor.getActiveCount());
}
//Look through the results of the tasks and get the next node that is closest by
currentNode = getNodeShortestDistanced();
//Reset the threadCounter for next iteration
this.setCount(0);
}
}
边数除以线程数。所以 8 个边和 2 个线程意味着每个线程将并行处理 4 个边。
public void calculateShortestDistances(int numberOfThreads) {
int threadCounter = this.getCount();
this.setCount(count + 1);
// Loop round the edges that are joined to the current node
currentNodeEdges = this.nodes[currentNode].getEdges();
int edgesPerThread = currentNodeEdges.size() / numberOfThreads;
int modulo = currentNodeEdges.size() % numberOfThreads;
this.nodes[0].setDistanceFromSource(0);
//Process the edges per thread
for (int joinedEdge = (edgesPerThread * threadCounter); joinedEdge < (edgesPerThread * (threadCounter + 1)); joinedEdge++) {
System.out.println("Start: " + (edgesPerThread * threadCounter) + ". End: " + (edgesPerThread * (threadCounter + 1) + ".JoinedEdge: " + joinedEdge) + ". Total: " + currentNodeEdges.size());
// Determine the joined edge neighbour of the current node
int neighbourIndex = currentNodeEdges.get(joinedEdge).getNeighbourIndex(currentNode);
// Only interested in an unvisited neighbour
if (!this.nodes[neighbourIndex].isVisited()) {
// Calculate the tentative distance for the neighbour
int tentative = this.nodes[currentNode].getDistanceFromSource() + currentNodeEdges.get(joinedEdge).getLength();
// Overwrite if the tentative distance is less than what's currently stored
if (tentative < nodes[neighbourIndex].getDistanceFromSource()) {
nodes[neighbourIndex].setDistanceFromSource(tentative);
}
}
}
//if we have a modulo above 0, the last thread will process the remaining edges
if (modulo > 0 && numberOfThreads == (threadCounter + 1)) {
for (int joinedEdge = (edgesPerThread * threadCounter); joinedEdge < (edgesPerThread * (threadCounter) + modulo); joinedEdge++) {
// Determine the joined edge neighbour of the current node
int neighbourIndex = currentNodeEdges.get(joinedEdge).getNeighbourIndex(currentNode);
// Only interested in an unvisited neighbour
if (!this.nodes[neighbourIndex].isVisited()) {
// Calculate the tentative distance for the neighbour
int tentative = this.nodes[currentNode].getDistanceFromSource() + currentNodeEdges.get(joinedEdge).getLength();
// Overwrite if the tentative distance is less than what's currently stored
if (tentative < nodes[neighbourIndex].getDistanceFromSource()) {
nodes[neighbourIndex].setDistanceFromSource(tentative);
}
}
}
}
// All neighbours are checked so this node is now visited
nodes[currentNode].setVisited(true);
}
谢谢你的帮助！
您应该查看 CyclicBarrier 或 CountDownLatch。这两个都允许您阻止线程启动，除非其他线程已发出它们已完成的信号。它们的区别在于CyclicBarrier是可重复使用的，即可以多次使用，而CountDownLatch是one-shot，不能重新计数。
来自 Javadocs 的解释：
A CountDownLatch is a synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
A CyclicBarrier is a synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point. CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. The barrier is called cyclic because it can be re-used after the waiting threads are released.
https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/CyclicBarrier.html
下面是一个使用 CountDownLatch 等待池中所有线程的简单演示：
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class WaitForAllThreadsInPool {
private static int MAX_CYCLES = 10;
public static void main(String args[]) throws InterruptedException, IOException {
new WaitForAllThreadsInPool().apply(4);
}
public void apply(int numberOfThreads) {
ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
CountDownLatch cdl = new CountDownLatch(numberOfThreads);
class DijkstraTask implements Runnable {
private final String name;
private final CountDownLatch cdl;
private final Random rnd = new Random();
public DijkstraTask(String name, CountDownLatch cdl) {
this.name = name;
this.cdl = cdl;
}
@Override
public void run() {
calculateShortestDistances(1+ rnd.nextInt(MAX_CYCLES), cdl, name);
}
}
for (int t = 0; t < numberOfThreads; t++) {
executor.execute(new DijkstraTask("Task " + t, cdl));
}
//wait for all threads to finish
try {
cdl.await();
System.out.println("-all done-");
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
public void calculateShortestDistances(int numberOfWorkCycles, CountDownLatch cdl, String name) {
//simulate long process
for(int cycle = 1 ; cycle <= numberOfWorkCycles; cycle++){
System.out.println(name + " cycle "+ cycle + "/"+ numberOfWorkCycles );
try {
TimeUnit.MILLISECONDS.sleep(1000);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
cdl.countDown(); //thread finished
}
}
输出样本：
Task 0 cycle 1/3
Task 1 cycle 1/2
Task 3 cycle 1/9
Task 2 cycle 1/3
Task 0 cycle 2/3
Task 1 cycle 2/2
Task 2 cycle 2/3
Task 3 cycle 2/9
Task 0 cycle 3/3
Task 2 cycle 3/3
Task 3 cycle 3/9
Task 3 cycle 4/9
Task 3 cycle 5/9
Task 3 cycle 6/9
Task 3 cycle 7/9
Task 3 cycle 8/9
Task 3 cycle 9/9
-all done-
您可以使用 invokeAll:
//Add task for each node
Collection<Callable<Object>> tasks = new ArrayList<>(numberOfThreads);
for (int t = 0; t < numberOfThreads; t++) {
tasks.add(Executors.callable(new DijkstraTask("Task " + t)));
}
//Wait until finished
executor.invokeAll(tasks);
我正在努力使 Dijkstra 算法并行化。使每个节点线程查看当前节点的所有边缘。这与线程并行，但开销太大。这导致比算法的顺序版本更长的时间。
添加了 ThreadPool 来解决这个问题，但我无法等待任务完成才能继续下一次迭代。只有完成一个节点的所有任务后，我们才能继续。在我可以按节点搜索下一个最近的任务之前，我们需要所有任务的结果。
我试过 executor.shutdown() 但使用这种方法它不会接受新任务。我们如何才能在循环中等待直到每个任务完成，而不必每次都声明 ThreadPoolExecutor。这样做会破坏通过使用它而不是常规线程来减少开销的目的。
我想到的一件事是添加任务（边缘）的 BlockingQueue。但是对于这个解决方案，我一直在等待任务在没有 shudown() 的情况下完成。
public void apply(int numberOfThreads) {
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(numberOfThreads);
class DijkstraTask implements Runnable {
private String name;
public DijkstraTask(String name) {
this.name = name;
}
public String getName() {
return name;
}
@Override
public void run() {
calculateShortestDistances(numberOfThreads);
}
}
// Visit every node, in order of stored distance
for (int i = 0; i < this.nodes.length; i++) {
//Add task for each node
for (int t = 0; t < numberOfThreads; t++) {
executor.execute(new DijkstraTask("Task " + t));
}
//Wait until finished?
while (executor.getActiveCount() > 0) {
System.out.println("Active count: " + executor.getActiveCount());
}
//Look through the results of the tasks and get the next node that is closest by
currentNode = getNodeShortestDistanced();
//Reset the threadCounter for next iteration
this.setCount(0);
}
}
边数除以线程数。所以 8 个边和 2 个线程意味着每个线程将并行处理 4 个边。
public void calculateShortestDistances(int numberOfThreads) {
int threadCounter = this.getCount();
this.setCount(count + 1);
// Loop round the edges that are joined to the current node
currentNodeEdges = this.nodes[currentNode].getEdges();
int edgesPerThread = currentNodeEdges.size() / numberOfThreads;
int modulo = currentNodeEdges.size() % numberOfThreads;
this.nodes[0].setDistanceFromSource(0);
//Process the edges per thread
for (int joinedEdge = (edgesPerThread * threadCounter); joinedEdge < (edgesPerThread * (threadCounter + 1)); joinedEdge++) {
System.out.println("Start: " + (edgesPerThread * threadCounter) + ". End: " + (edgesPerThread * (threadCounter + 1) + ".JoinedEdge: " + joinedEdge) + ". Total: " + currentNodeEdges.size());
// Determine the joined edge neighbour of the current node
int neighbourIndex = currentNodeEdges.get(joinedEdge).getNeighbourIndex(currentNode);
// Only interested in an unvisited neighbour
if (!this.nodes[neighbourIndex].isVisited()) {
// Calculate the tentative distance for the neighbour
int tentative = this.nodes[currentNode].getDistanceFromSource() + currentNodeEdges.get(joinedEdge).getLength();
// Overwrite if the tentative distance is less than what's currently stored
if (tentative < nodes[neighbourIndex].getDistanceFromSource()) {
nodes[neighbourIndex].setDistanceFromSource(tentative);
}
}
}
//if we have a modulo above 0, the last thread will process the remaining edges
if (modulo > 0 && numberOfThreads == (threadCounter + 1)) {
for (int joinedEdge = (edgesPerThread * threadCounter); joinedEdge < (edgesPerThread * (threadCounter) + modulo); joinedEdge++) {
// Determine the joined edge neighbour of the current node
int neighbourIndex = currentNodeEdges.get(joinedEdge).getNeighbourIndex(currentNode);
// Only interested in an unvisited neighbour
if (!this.nodes[neighbourIndex].isVisited()) {
// Calculate the tentative distance for the neighbour
int tentative = this.nodes[currentNode].getDistanceFromSource() + currentNodeEdges.get(joinedEdge).getLength();
// Overwrite if the tentative distance is less than what's currently stored
if (tentative < nodes[neighbourIndex].getDistanceFromSource()) {
nodes[neighbourIndex].setDistanceFromSource(tentative);
}
}
}
}
// All neighbours are checked so this node is now visited
nodes[currentNode].setVisited(true);
}
谢谢你的帮助！
您应该查看 CyclicBarrier 或 CountDownLatch。这两个都允许您阻止线程启动，除非其他线程已发出它们已完成的信号。它们的区别在于CyclicBarrier是可重复使用的，即可以多次使用，而CountDownLatch是one-shot，不能重新计数。
来自 Javadocs 的解释：
A CountDownLatch is a synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
A CyclicBarrier is a synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point. CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. The barrier is called cyclic because it can be re-used after the waiting threads are released.
https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/CyclicBarrier.html
下面是一个使用 CountDownLatch 等待池中所有线程的简单演示：
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class WaitForAllThreadsInPool {
private static int MAX_CYCLES = 10;
public static void main(String args[]) throws InterruptedException, IOException {
new WaitForAllThreadsInPool().apply(4);
}
public void apply(int numberOfThreads) {
ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
CountDownLatch cdl = new CountDownLatch(numberOfThreads);
class DijkstraTask implements Runnable {
private final String name;
private final CountDownLatch cdl;
private final Random rnd = new Random();
public DijkstraTask(String name, CountDownLatch cdl) {
this.name = name;
this.cdl = cdl;
}
@Override
public void run() {
calculateShortestDistances(1+ rnd.nextInt(MAX_CYCLES), cdl, name);
}
}
for (int t = 0; t < numberOfThreads; t++) {
executor.execute(new DijkstraTask("Task " + t, cdl));
}
//wait for all threads to finish
try {
cdl.await();
System.out.println("-all done-");
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
public void calculateShortestDistances(int numberOfWorkCycles, CountDownLatch cdl, String name) {
//simulate long process
for(int cycle = 1 ; cycle <= numberOfWorkCycles; cycle++){
System.out.println(name + " cycle "+ cycle + "/"+ numberOfWorkCycles );
try {
TimeUnit.MILLISECONDS.sleep(1000);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
cdl.countDown(); //thread finished
}
}
输出样本：
Task 0 cycle 1/3
Task 1 cycle 1/2
Task 3 cycle 1/9
Task 2 cycle 1/3
Task 0 cycle 2/3
Task 1 cycle 2/2
Task 2 cycle 2/3
Task 3 cycle 2/9
Task 0 cycle 3/3
Task 2 cycle 3/3
Task 3 cycle 3/9
Task 3 cycle 4/9
Task 3 cycle 5/9
Task 3 cycle 6/9
Task 3 cycle 7/9
Task 3 cycle 8/9
Task 3 cycle 9/9
-all done-
您可以使用 invokeAll:
//Add task for each node
Collection<Callable<Object>> tasks = new ArrayList<>(numberOfThreads);
for (int t = 0; t < numberOfThreads; t++) {
tasks.add(Executors.callable(new DijkstraTask("Task " + t)));
}
//Wait until finished
executor.invokeAll(tasks);