什么决定了Java ForkJoinPool创建的线程数？

java parallel-processing threadpool fork-join

2022-08-31 20:20:42

据我所知，该池会创建固定数量的线程（默认值：内核数），并且永远不会创建更多线程（除非应用程序通过使用指示需要这些线程）。ForkJoinPoolmanagedBlock

但是，使用我发现在创建30，000个任务的程序中（），执行这些任务平均使用700个线程（每次创建任务时计数线程）。任务不做I / O，而是纯计算;唯一的任务间同步是调用和访问 s，即没有线程阻塞操作。ForkJoinPool.getPoolSize()RecursiveActionForkJoinPoolForkJoinTask.join()AtomicBoolean

由于不阻止我所理解的调用线程，因此没有理由阻止池中的任何线程，因此（我假设）应该没有理由创建任何进一步的线程（这显然正在发生）。join()

那么，为什么要创建这么多线程呢？哪些因素决定了创建的线程数？ForkJoinPool

我本来希望这个问题可以在不发布代码的情况下得到回答，但在这里它是根据要求提出的。此代码是从四倍大小的程序中摘录的，缩小到基本部分;它不会按原样编译。如果需要，我当然也可以发布完整的程序。

该程序使用深度优先搜索在迷宫中搜索从给定起点到给定终点的路径。解决方案保证存在。A从某个给定点开始，并继续从当前点到达的所有相邻点。它不是在每个分支点创建一个新的分支点（这将创建太多任务），而是将除一个邻居之外的所有邻居都推送到回溯堆栈上，以便稍后处理，并且只继续一个未推送到堆栈的邻居。一旦它以这种方式到达死胡同，最近推到回溯堆栈的点就会弹出，并且从那里继续搜索（相应地从taks的起点减少构建的路径）。一旦任务发现其回溯堆栈大于某个阈值，就会创建新任务;从那时起，任务虽然继续从其回溯堆栈中弹出，直到耗尽，但在到达分支点时不会将任何进一步的点推送到其堆栈，而是为每个此类点创建一个新任务。因此，可以使用堆栈限制阈值调整任务的大小。compute()SolverTaskRecursiveActionSolverTask

我上面引用的数字（“30，000个任务，平均700个线程”）来自搜索5000x5000个单元格的迷宫。所以，这是基本代码：

class SolverTask extends RecursiveTask<ArrayDeque<Point>> {
// Once the backtrack stack has reached this size, the current task
// will never add another cell to it, but create a new task for each
// newly discovered branch:
private static final int MAX_BACKTRACK_CELLS = 100*1000;

/**
 * @return Tries to compute a path through the maze from local start to end
 * and returns that (or null if no such path found)
 */
@Override
public ArrayDeque<Point>  compute() {
    // Is this task still accepting new branches for processing on its own,
    // or will it create new tasks to handle those?
    boolean stillAcceptingNewBranches = true;
    Point current = localStart;
    ArrayDeque<Point> pathFromLocalStart = new ArrayDeque<Point>();  // Path from localStart to (including) current
    ArrayDeque<PointAndDirection> backtrackStack = new ArrayDeque<PointAndDirection>();
    // Used as a stack: Branches not yet taken; solver will backtrack to these branching points later

    Direction[] allDirections = Direction.values();

    while (!current.equals(end)) {
        pathFromLocalStart.addLast(current);
        // Collect current's unvisited neighbors in random order: 
        ArrayDeque<PointAndDirection> neighborsToVisit = new ArrayDeque<PointAndDirection>(allDirections.length);  
        for (Direction directionToNeighbor: allDirections) {
            Point neighbor = current.getNeighbor(directionToNeighbor);

            // contains() and hasPassage() are read-only methods and thus need no synchronization
            if (maze.contains(neighbor) && maze.hasPassage(current, neighbor) && maze.visit(neighbor))
                neighborsToVisit.add(new PointAndDirection(neighbor, directionToNeighbor.opposite));
        }
        // Process unvisited neighbors
        if (neighborsToVisit.size() == 1) {
            // Current node is no branch: Continue with that neighbor
            current = neighborsToVisit.getFirst().getPoint();
            continue;
        }
        if (neighborsToVisit.size() >= 2) {
            // Current node is a branch
            if (stillAcceptingNewBranches) {
                current = neighborsToVisit.removeLast().getPoint();
                // Push all neighbors except one on the backtrack stack for later processing
                for(PointAndDirection neighborAndDirection: neighborsToVisit) 
                    backtrackStack.push(neighborAndDirection);
                if (backtrackStack.size() > MAX_BACKTRACK_CELLS)
                    stillAcceptingNewBranches = false;
                // Continue with the one neighbor that was not pushed onto the backtrack stack
                continue;
            } else {
                // Current node is a branch point, but this task does not accept new branches any more: 
                // Create new task for each neighbor to visit and wait for the end of those tasks
                SolverTask[] subTasks = new SolverTask[neighborsToVisit.size()];
                int t = 0;
                for(PointAndDirection neighborAndDirection: neighborsToVisit)  {
                    SolverTask task = new SolverTask(neighborAndDirection.getPoint(), end, maze);
                    task.fork();
                    subTasks[t++] = task;
                }
                for (SolverTask task: subTasks) {
                    ArrayDeque<Point> subTaskResult = null;
                    try {
                        subTaskResult = task.join();
                    } catch (CancellationException e) {
                        // Nothing to do here: Another task has found the solution and cancelled all other tasks
                    }
                    catch (Exception e) {
                        e.printStackTrace();
                    }
                    if (subTaskResult != null) { // subtask found solution
                        pathFromLocalStart.addAll(subTaskResult);
                        // No need to wait for the other subtasks once a solution has been found
                        return pathFromLocalStart;
                    }
                } // for subTasks
            } // else (not accepting any more branches) 
        } // if (current node is a branch)
        // Current node is dead end or all its neighbors lead to dead ends:
        // Continue with a node from the backtracking stack, if any is left:
        if (backtrackStack.isEmpty()) {
            return null; // No more backtracking avaible: No solution exists => end of this task
        }
        // Backtrack: Continue with cell saved at latest branching point:
        PointAndDirection pd = backtrackStack.pop();
        current = pd.getPoint();
        Point branchingPoint = current.getNeighbor(pd.getDirectionToBranchingPoint());
        // DEBUG System.out.println("Backtracking to " +  branchingPoint);
        // Remove the dead end from the top of pathSoFar, i.e. all cells after branchingPoint:
        while (!pathFromLocalStart.peekLast().equals(branchingPoint)) {
            // DEBUG System.out.println("    Going back before " + pathSoFar.peekLast());
            pathFromLocalStart.removeLast();
        }
        // continue while loop with newly popped current
    } // while (current ...
    if (!current.equals(end)) {         
        // this task was interrupted by another one that already found the solution 
        // and should end now therefore:
        return null;
    } else {
        // Found the solution path:
        pathFromLocalStart.addLast(current);
        return pathFromLocalStart;
    }
} // compute()
} // class SolverTask

@SuppressWarnings("serial")
public class ParallelMaze  {

// for each cell in the maze: Has the solver visited it yet?
private final AtomicBoolean[][] visited;

/**
 * Atomically marks this point as visited unless visited before
 * @return whether the point was visited for the first time, i.e. whether it could be marked
 */
boolean visit(Point p) {
    return  visited[p.getX()][p.getY()].compareAndSet(false, true);
}

public static void main(String[] args) {
    ForkJoinPool pool = new ForkJoinPool();
    ParallelMaze maze = new ParallelMaze(width, height, new Point(width-1, 0), new Point(0, height-1));
    // Start initial task
    long startTime = System.currentTimeMillis();
     // since SolverTask.compute() expects its starting point already visited, 
    // must do that explicitly for the global starting point:
    maze.visit(maze.start);
    maze.solution = pool.invoke(new SolverTask(maze.start, maze.end, maze));
    // One solution is enough: Stop all tasks that are still running
    pool.shutdownNow();
    pool.awaitTermination(Integer.MAX_VALUE, TimeUnit.DAYS);
    long endTime = System.currentTimeMillis();
    System.out.println("Computed solution of length " + maze.solution.size() + " to maze of size " + 
            width + "x" + height + " in " + ((float)(endTime - startTime))/1000 + "s.");
}

答案 1

关于堆栈溢出有相关的问题：

ForkJoinPool 在调用All/join 期间停止

ForkJoinPool似乎浪费了一个线程

我做了一个可运行的精简版本（我使用的jvm参数：-Xms256m -Xmx1024m -Xss8m）：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;

public class Test1 {

    private static ForkJoinPool pool = new ForkJoinPool(2);

    private static class SomeAction extends RecursiveAction {

        private int counter;         //recursive counter
        private int childrenCount=80;//amount of children to spawn
        private int idx;             // just for displaying

        private SomeAction(int counter, int idx) {
            this.counter = counter;
            this.idx = idx;
        }

        @Override
        protected void compute() {

            System.out.println(
                "counter=" + counter + "." + idx +
                " activeThreads=" + pool.getActiveThreadCount() +
                " runningThreads=" + pool.getRunningThreadCount() +
                " poolSize=" + pool.getPoolSize() +
                " queuedTasks=" + pool.getQueuedTaskCount() +
                " queuedSubmissions=" + pool.getQueuedSubmissionCount() +
                " parallelism=" + pool.getParallelism() +
                " stealCount=" + pool.getStealCount());
            if (counter <= 0) return;

            List<SomeAction> list = new ArrayList<>(childrenCount);
            for (int i=0;i<childrenCount;i++){
                SomeAction next = new SomeAction(counter-1,i);
                list.add(next);
                next.fork();
            }


            for (SomeAction action:list){
                action.join();
            }
        }
    }

    public static void main(String[] args) throws Exception{
        pool.invoke(new SomeAction(2,0));
    }
}

显然，当您执行联接时，当前线程会看到所需的任务尚未完成，并需要另一个任务自己完成。

它发生在 .java.util.concurrent.ForkJoinWorkerThread#joinTask

但是，此新任务生成了更多相同的任务，但它们无法在池中找到线程，因为线程被锁定在连接中。由于它无法知道释放它们需要多少时间（线程可能处于无限循环或永远处于死锁状态），因此会生成新线程（补偿Louis Wasserman提到的连接线程）：java.util.concurrent.ForkJoinPool#signalWork

因此，为了防止这种情况，您需要避免递归生成任务。

例如，如果在上面的代码中将初始参数设置为 1，则即使您将 childrenCount 增加十倍，活动线程量也将为 2。

另请注意，当活动线程数增加时，正在运行的线程数小于或等于并行度。

答案 2

从源注释：

补偿：除非已经有足够的活动线程，否则方法 tryPreBlock（）可能会创建或重新激活一个备用线程来补偿阻塞的 joiners，直到它们解除阻塞。

我认为正在发生的事情是，您没有很快完成任何任务，并且由于在提交新任务时没有可用的工作线程，因此创建了一个新线程。