死磕 java线程系列之ForkJoinPool深入解析

喜欢ヅ旅行 2023-05-30 09:13 100阅读 0赞

forkjoinpool

（手机横屏看源码更方便）

#

注：java源码分析部分如无特殊说明均基于 java8 版本。

注：本文基于ForkJoinPool分治线程池类。

简介

随着在硬件上多核处理器的发展和广泛使用，并发编程成为程序员必须掌握的一门技术，在面试中也经常考查面试者并发相关的知识。

今天，我们就来看一道面试题：

如何充分利用多核CPU，计算很大数组中所有整数的和？

剖析

  • 单线程相加？

我们最容易想到就是单线程相加，一个for循环搞定。

  • 线程池相加？

如果进一步优化，我们会自然而然地想到使用线程池来分段相加，最后再把每个段的结果相加。

  • 其它？

Yes，就是我们今天的主角——ForkJoinPool，但是它要怎么实现呢？似乎没怎么用过哈^^

三种实现

OK，剖析完了，我们直接来看三种实现，不墨迹，直接上菜。

  1. /**
  2.  * 计算1亿个整数的和
  3.  */
  4. public class ForkJoinPoolTest01 {
  5.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  6.         // 构造数据
  7.         int length = 100000000;
  8.         long[] arr = new long[length];
  9.         for (int i = 0; i < length; i  ) {
  10.             arr[i] = ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE);
  11.         }
  12.         // 单线程
  13.         singleThreadSum(arr);
  14.         // ThreadPoolExecutor线程池
  15.         multiThreadSum(arr);
  16.         // ForkJoinPool线程池
  17.         forkJoinSum(arr);
  19.     }
  21.     private static void singleThreadSum(long[] arr) {
  22.         long start = System.currentTimeMillis();
  24.         long sum = 0;
  25.         for (int i = 0; i < arr.length; i  ) {
  26.             // 模拟耗时，本文由公从号“彤哥读源码”原创
  27.             sum  = (arr[i]/3*3/3*3/3*3/3*3/3*3);
  28.         }
  30.         System.out.println("sum: "   sum);
  31.         System.out.println("single thread elapse: "   (System.currentTimeMillis() - start));
  33.     }
  35.     private static void multiThreadSum(long[] arr) throws ExecutionException, InterruptedException {
  36.         long start = System.currentTimeMillis();
  38.         int count = 8;
  39.         ExecutorService threadPool = Executors.newFixedThreadPool(count);
  40.         List<Future<Long>> list = new ArrayList<>();
  41.         for (int i = 0; i < count; i  ) {
  42.             int num = i;
  43.             // 分段提交任务
  44.             Future<Long> future = threadPool.submit(() -> {
  45.                 long sum = 0;
  46.                 for (int j = arr.length / count * num; j < (arr.length / count * (num   1)); j  ) {
  47.                     try {
  48.                         // 模拟耗时
  49.                         sum  = (arr[j]/3*3/3*3/3*3/3*3/3*3);
  50.                     } catch (Exception e) {
  51.                         e.printStackTrace();
  52.                     }
  53.                 }
  54.                 return sum;
  55.             });
  56.             list.add(future);
  57.         }
  59.         // 每个段结果相加
  60.         long sum = 0;
  61.         for (Future<Long> future : list) {
  62.             sum  = future.get();
  63.         }
  65.         System.out.println("sum: "   sum);
  66.         System.out.println("multi thread elapse: "   (System.currentTimeMillis() - start));
  67.     }
  69.     private static void forkJoinSum(long[] arr) throws ExecutionException, InterruptedException {
  70.         long start = System.currentTimeMillis();
  72.         ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
  73.         // 提交任务
  74.         ForkJoinTask<Long> forkJoinTask = forkJoinPool.submit(new SumTask(arr, 0, arr.length));
  75.         // 获取结果
  76.         Long sum = forkJoinTask.get();
  78.         forkJoinPool.shutdown();
  80.         System.out.println("sum: "   sum);
  81.         System.out.println("fork join elapse: "   (System.currentTimeMillis() - start));
  82.     }
  84.     private static class SumTask extends RecursiveTask<Long> {
  85.         private long[] arr;
  86.         private int from;
  87.         private int to;
  89.         public SumTask(long[] arr, int from, int to) {
  90.             this.arr = arr;
  91.             this.from = from;
  92.             this.to = to;
  93.         }
  95.         @Override
  96.         protected Long compute() {
  97.             // 小于1000的时候直接相加，可灵活调整
  98.             if (to - from <= 1000) {
  99.                 long sum = 0;
  100.                 for (int i = from; i < to; i  ) {
  101.                     // 模拟耗时
  102.                     sum  = (arr[i]/3*3/3*3/3*3/3*3/3*3);
  103.                 }
  104.                 return sum;
  105.             }
  107.             // 分成两段任务，本文由公从号“彤哥读源码”原创
  108.             int middle = (from   to) / 2;
  109.             SumTask left = new SumTask(arr, from, middle);
  110.             SumTask right = new SumTask(arr, middle, to);
  112.             // 提交左边的任务
  113.             left.fork();
  114.             // 右边的任务直接利用当前线程计算，节约开销
  115.             Long rightResult = right.compute();
  116.             // 等待左边计算完毕
  117.             Long leftResult = left.join();
  118.             // 返回结果
  119.             return leftResult   rightResult;
  120.         }
  121.     }
  122. }

彤哥偷偷地告诉你，实际上计算1亿个整数相加，单线程是最快的，我的电脑大概是100ms左右，使用线程池反而会变慢。

所以，为了演示ForkJoinPool的牛逼之处，我把每个数都/3*3/3*3/3*3/3*3/3*3了一顿操作，用来模拟计算耗时。

来看结果：

  1. sum: 107352457433800662
  2. single thread elapse: 789
  3. sum: 107352457433800662
  4. multi thread elapse: 228
  5. sum: 107352457433800662
  6. fork join elapse: 189

可以看到，ForkJoinPool相对普通线程池还是有很大提升的。

问题：普通线程池能否实现ForkJoinPool这种计算方式呢，即大任务拆中任务，中任务拆小任务，最后再汇总？

forkjoinpool

你可以试试看(-᷅_-᷄)

OK，下面我们正式进入ForkJoinPool的解析。

分治法

  • 基本思想

把一个规模大的问题划分为规模较小的子问题，然后分而治之，最后合并子问题的解得到原问题的解。

  • 步骤

（1）分割原问题：

（2）求解子问题：

（3）合并子问题的解为原问题的解。

在分治法中，子问题一般是相互独立的，因此，经常通过递归调用算法来求解子问题。

  • 典型应用场景

（1）二分搜索

（2）大整数乘法

（3）Strassen矩阵乘法

（4）棋盘覆盖

（5）归并排序

（6）快速排序

（7）线性时间选择

（8）汉诺塔

ForkJoinPool继承体系

ForkJoinPool是 java 7 中新增的线程池类，它的继承体系如下：

forkjoinpool

ForkJoinPool和ThreadPoolExecutor都是继承自AbstractExecutorService抽象类，所以它和ThreadPoolExecutor的使用几乎没有多少区别，除了任务变成了ForkJoinTask以外。

这里又运用到了一种很重要的设计原则——开闭原则——对修改关闭，对扩展开放。

可见整个线程池体系一开始的接口设计就很好，新增一个线程池类，不会对原有的代码造成干扰，还能利用原有的特性。

ForkJoinTask

两个主要方法

  • fork()

fork()方法类似于线程的Thread.start()方法，但是它不是真的启动一个线程，而是将任务放入到工作队列中。

  • join()

join()方法类似于线程的Thread.join()方法，但是它不是简单地阻塞线程，而是利用工作线程运行其它任务。当一个工作线程中调用了join()方法，它将处理其它任务，直到注意到目标子任务已经完成了。

三个子类

  • RecursiveAction

无返回值任务。

  • RecursiveTask

有返回值任务。

  • CountedCompleter

无返回值任务，完成任务后可以触发回调。

ForkJoinPool内部原理

ForkJoinPool内部使用的是“工作窃取”算法实现的。

forkjoinpool

（1）每个工作线程都有自己的工作队列WorkQueue；

（2）这是一个双端队列，它是线程私有的；

（3）ForkJoinTask中fork的子任务，将放入运行该任务的工作线程的队头，工作线程将以LIFO的顺序来处理工作队列中的任务；

（4）为了最大化地利用CPU，空闲的线程将从其它线程的队列中“窃取”任务来执行；

（5）从工作队列的尾部窃取任务，以减少竞争；

（6）双端队列的操作：push()/pop()仅在其所有者工作线程中调用，poll()是由其它线程窃取任务时调用的；

（7）当只剩下最后一个任务时，还是会存在竞争，是通过CAS来实现的；

forkjoinpool

ForkJoinPool最佳实践

（1）最适合的是计算密集型任务，本文由公从号“彤哥读源码”原创；

（2）在需要阻塞工作线程时，可以使用ManagedBlocker；

（3）不应该在RecursiveTask 的内部使用ForkJoinPool.invoke()/invokeAll()；

总结

（1）ForkJoinPool特别适合于“分而治之”算法的实现；

（2）ForkJoinPool和ThreadPoolExecutor是互补的，不是谁替代谁的关系，二者适用的场景不同；

（3）ForkJoinTask有两个核心方法——fork()和join()，有三个重要子类——RecursiveAction、RecursiveTask和CountedCompleter；

（4）ForkjoinPool内部基于“工作窃取”算法实现；

（5）每个线程有自己的工作队列，它是一个双端队列，自己从队列头存取任务，其它线程从尾部窃取任务；

（6）ForkJoinPool最适合于计算密集型任务，但也可以使用ManagedBlocker以便用于阻塞型任务；

（7）RecursiveTask内部可以少调用一次fork()，利用当前线程处理，这是一种技巧；

彩蛋

ManagedBlocker怎么使用？

答：ManagedBlocker相当于明确告诉ForkJoinPool框架要阻塞了，ForkJoinPool就会启另一个线程来运行任务，以最大化地利用CPU。

请看下面的例子，自己琢磨哈^^。

  1. /**
  2.  * 斐波那契数列
  3.  * 一个数是它前面两个数之和
  4.  * 1,1,2,3,5,8,13,21
  5.  */
  6. public class Fibonacci {
  8.     public static void main(String[] args) {
  9.         long time = System.currentTimeMillis();
  10.         Fibonacci fib = new Fibonacci();
  11.         int result = fib.f(1_000).bitCount();
  12.         time = System.currentTimeMillis() - time;
  13.         System.out.println("result，本文由公从号“彤哥读源码”原创 = "   result);
  14.         System.out.println("test1_000() time = "   time);
  15.     }
  17.     public BigInteger f(int n) {
  18.         Map<Integer, BigInteger> cache = new ConcurrentHashMap<>();
  19.         cache.put(0, BigInteger.ZERO);
  20.         cache.put(1, BigInteger.ONE);
  21.         return f(n, cache);
  22.     }
  24.     private final BigInteger RESERVED = BigInteger.valueOf(-1000);
  26.     public BigInteger f(int n, Map<Integer, BigInteger> cache) {
  27.         BigInteger result = cache.putIfAbsent(n, RESERVED);
  28.         if (result == null) {
  30.             int half = (n   1) / 2;
  32.             RecursiveTask<BigInteger> f0_task = new RecursiveTask<BigInteger>() {
  33.                 @Override
  34.                 protected BigInteger compute() {
  35.                     return f(half - 1, cache);
  36.                 }
  37.             };
  38.             f0_task.fork();
  40.             BigInteger f1 = f(half, cache);
  41.             BigInteger f0 = f0_task.join();
  43.             long time = n > 10_000 ? System.currentTimeMillis() : 0;
  44.             try {
  46.                 if (n % 2 == 1) {
  47.                     result = f0.multiply(f0).add(f1.multiply(f1));
  48.                 } else {
  49.                     result = f0.shiftLeft(1).add(f1).multiply(f1);
  50.                 }
  51.                 synchronized (RESERVED) {
  52.                     cache.put(n, result);
  53.                     RESERVED.notifyAll();
  54.                 }
  55.             } finally {
  56.                 time = n > 10_000 ? System.currentTimeMillis() - time : 0;
  57.                 if (time > 50)
  58.                     System.out.printf("f(%d) took %d%n", n, time);
  59.             }
  60.         } else if (result == RESERVED) {
  61.             try {
  62.                 ReservedFibonacciBlocker blocker = new ReservedFibonacciBlocker(n, cache);
  63.                 ForkJoinPool.managedBlock(blocker);
  64.                 result = blocker.result;
  65.             } catch (InterruptedException e) {
  66.                 throw new CancellationException("interrupted");
  67.             }
  69.         }
  70.         return result;
  71.         // return f(n - 1).add(f(n - 2));
  72.     }
  74.     private class ReservedFibonacciBlocker implements ForkJoinPool.ManagedBlocker {
  75.         private BigInteger result;
  76.         private final int n;
  77.         private final Map<Integer, BigInteger> cache;
  79.         public ReservedFibonacciBlocker(int n, Map<Integer, BigInteger> cache) {
  80.             this.n = n;
  81.             this.cache = cache;
  82.         }
  84.         @Override
  85.         public boolean block() throws InterruptedException {
  86.             synchronized (RESERVED) {
  87.                 while (!isReleasable()) {
  88.                     RESERVED.wait();
  89.                 }
  90.             }
  91.             return true;
  92.         }
  94.         @Override
  95.         public boolean isReleasable() {
  96.             return (result = cache.get(n)) != RESERVED;
  97.         }
  98.     }
  99. }

#

欢迎关注我的公众号“彤哥读源码”，查看更多源码系列文章, 与彤哥一起畅游源码的海洋。

qrcode

发表评论

表情：
评论列表 （有 0 条评论，100人围观）

还没有评论，来说两句吧...

相关阅读