六、Java并发 Java ForkJoinPool

文章永久连接：https://tech.souyunku.com/?p=5575

ForkJoinPool 是Java 7 中引入的 fork/join 框架的核心之一。它解决了一个常见的问题： 如何在递归中生成多个任务。因为，即使是使用一个简单的 ThreadPoolExecutor ，也会在不断的递归中快速耗尽线程。因为每个任务或子任务都需要自己的线程来运行。

在 fork/join 框架中，任何任务都可以生成 ( fork ) 多个子任务并使用 join() 方法等待它们的完成。fork/join 框架的好处是它不会为每个任务或子任务创建新线程，而是实现了 工作窃取 ( Work Stealing ) 算法。关于 fork/join 框架的详细信息，你可以访问我们的 一文秒懂 Java Fork/Join。

接下来，我们看一个使用 ForkJoinPool 遍历节点树并计算所有叶值之和的简单示例。在这个示例中，树是一个由节点，int 值和一组子节点组成。

static class TreeNode {

    int value;

    Set<TreeNode> children;

    TreeNode(int value, TreeNode... children) {
        this.value = value;
        this.children = Sets.newHashSet(children);
    }
}

创建了树 TreeNode 之后，如果我们想要并行地对树中的所有值求和，我们需要实现一个 RecursiveTask<Integer> 接口。每个任务都接收自己的节点，并将其值添加到其子节点的值之和上。

要计算子节点值的总和，任务实现执行以下操作

1、 将子节点集合转换为流 ( stream )
2、 映射前面操作中创建的流，为每个元素创建一个新的 CountingTask
3、 通过 fork 执行每个子任务
4、 通过在每个 fork 任务上调用 join() 方法来收集结果
5、 使用 ·Collectors.summingInt` 收集器对结果求和

public static class CountingTask extends RecursiveTask<Integer> {

    private final TreeNode node;

    public CountingTask(TreeNode node) {
        this.node = node;
    }

    @Override
    protected Integer compute() {
        return node.value + node.children.stream()
          .map(childNode -> new CountingTask(childNode).fork())
          .collect(Collectors.summingInt(ForkJoinTask::join));
    }
}

在树上运行计算的代码非常简单：

TreeNode tree = new TreeNode(5,
  new TreeNode(3), new TreeNode(2,
    new TreeNode(2), new TreeNode(8)));

ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
int sum = forkJoinPool.invoke(new CountingTask(tree));

干货推荐

本站推荐：精选优质专栏

附录：Java 并发：系列文章

• 一、Java并发 Java ExecutorService
• 二、Java并发 Java Fork-Join
• 三、Java并发 Java 线程池 ( Thread Pool ) （上）
• 四、Java并发 Java 线程池之 ThreadPoolExecutor
• 五、Java并发 ScheduledThreadPoolExecutor
• 【当前读到】六、Java并发 Java ForkJoinPool
• 七、Java并发 Java Google Guava 实现
• 八、Java并发 Java CompletableFuture ( 上 )
• 九、Java并发 Java CompletableFuture ( 下 )
• 十、Java并发系列之 Java 并发编程面试题
• 十一、Java并发 Java CountDownLatch
• 十二、Java并发 Java BlockingQueue
• 十三、Java并发 Java java.util.concurrent.Locks
• 十四、Java并发 Java 守护线程 ( Daemon Thread )
• 十五、Java并发 Java java.util.concurrent.Future
• 十六、Java并发 Java ThreadLocalRandom
• 十七、Java并发 Java Thread 生命周期
• 十八、Java并发 Java 之 Runnable 还是 Thread ?
• 十九、Java并发 Java wait() 和 notify() 方法