java并发 java7之fork-join框架

曾经终败给现在 2022-08-11 00:40 103阅读 0赞

如果让我个人理解什么是fork-join，我立刻会想到hadoop的map/reduce。他们是同一种模型。更早之前，笔者就看过关于fork-join的相关文章，但是理解的都不够深刻。自从深入研究了java多线程这块的相关技术，再回头来看fork-join，觉得真是个了不起的技术。  
  
**概述**  
fork/join框架是ExecutorService接口的一种具体实现，目的是为了帮助你更好地利用多处理器带来的好处。它是为那些能够被递归地拆解成子任务的工作类型量身设计的。其目的在于能够使用所有可用的运算能力来提升你的应用的性能。    
类似于ExecutorService接口的其他实现，fork/join框架会将任务分发给线程池中的工作线程。fork/join框架的独特之处在与它使用工作窃取(work-stealing)算法。完成自己的工作而处于空闲的工作线程能够从其他仍然处于忙碌(busy)状态的工作线程处窃取等待执行的任务。 fork/join框架的核心是ForkJoinPool类，它是对AbstractExecutorService类的扩展。ForkJoinPool实现了工作偷取算法，并可以执行ForkJoinTask任务。  
  
**Divide and conquer**  
合并排序是 divide-and-conquer 算法的一个例子，在这种算法中将一个问题递归分解成子问题，再将子问题的解决方案组合得到最终结果。 divide-and-conquer 算法也可用于顺序环境中，但是在并行环境中更加有效，因为可以并行处理子问题。  
并行 divide-and-conquer 算法首先对问题进行评估，确定其大小是否更适合使用顺序解决方案；通常，可通过将问题大小与某个阙值进行比较完成。如果问题大到需要并行分解，算法会将其分解成两个或更多子问题，并以并行方式对子问题递归调用算法本身，然后等待子问题的结果，最后合并这些结果。用于选择顺序和并行执行方法的理想阙值是协调并行任务的成本。如果协调成本为 0，更多的更细粒度的任务会提供更好的并行性；在需要转向顺序方法之前，协调成本越低，就可以划分更细粒度的任务。  
  
**递归（旧的方法）**  
如果从一个数组中，选一个最大值。我一般都会采用递归调用。这样的话只有当前线程的一个方法是运行的，其他方法都阻塞在线程栈中。所以在多核情况下，其他CPU都是空闲，没有得到充分利用。  
  
**并行计算（fork-join）**  
  
类似hadoop的map/reduce，可以将任务拆分成多个块，然后最终将这些子结果集合并成最终结果集。它的行为表现为当前任务是暂停的，并行执行两个子任务，而当前任务等待两个子任务的完成。然后就可以将两个子任务的结果进行合并。这种并行分解方法常常称作 fork-join，因为执行一个任务将首先分解（fork）为多个子任务，然后再合并（join）（完成后）。  
  
使用fork/join框架的第一步是编写执行一部分工作的代码。你的代码结构看起来应该与下面所示的伪代码类似：

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  if (当前这个任务工作量足够小)  
2.      直接完成这个任务  
3.  else  
4.      将这个任务或这部分工作分解成两个部分  
5.      分别触发(invoke)这两个子任务的执行，并等待结果

if (当前这个任务工作量足够小)
        直接完成这个任务
    else
        将这个任务或这部分工作分解成两个部分
        分别触发(invoke)这两个子任务的执行，并等待结果

示例代码如下：求

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  import java.util.concurrent.ForkJoinPool;  
2.  import java.util.concurrent.RecursiveTask;  
3.    
4.  /\*\* 
5.   \* @author piaohailin 
6.   \* @date   2014-3-26 
7.  \*/  
8.  public class Fibonacci extends RecursiveTask<Long> \{  
9.      private static final long serialVersionUID = 7875142223684511653L;  
10.     private final long        n;  
11.   
12.     Fibonacci(long n) \{  
13.         this.n = n;  
14.     \}  
15.   
16.     protected Long compute() \{  
17.         if (n <= 1) \{  
18.             return n;  
19.         \}  
20.         Fibonacci f1 = new Fibonacci(n - 1);  
21.         f1.fork();  
22.         Fibonacci f2 = new Fibonacci(n - 2);  
23.         return f2.compute() + f1.join();  
24.     \}  
25.   
26.     public static void main(String\[\] args) \{  
27.         Fibonacci task = new Fibonacci(10);  
28.         ForkJoinPool pool = new ForkJoinPool(4);  
29.   
30.         pool.invoke(task);  
31.         System.out.println(task.getRawResult());  
32.     \}  
33.   
34. \}

import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.RecursiveTask;
    
    /**
     * @author piaohailin
     * @date   2014-3-26
    */
    public class Fibonacci extends RecursiveTask<Long> {
        private static final long serialVersionUID = 7875142223684511653L;
        private final long        n;
    
        Fibonacci(long n) {
            this.n = n;
        }
    
        protected Long compute() {
            if (n <= 1) {
                return n;
            }
            Fibonacci f1 = new Fibonacci(n - 1);
            f1.fork();
            Fibonacci f2 = new Fibonacci(n - 2);
            return f2.compute() + f1.join();
        }
    
        public static void main(String[] args) {
            Fibonacci task = new Fibonacci(10);
            ForkJoinPool pool = new ForkJoinPool(4);
    
            pool.invoke(task);
            System.out.println(task.getRawResult());
        }
    
    }

**工作窃取**  
fork-join 框架通过一种称作工作窃取（work stealing） 的技术减少了工作队列的争用情况。每个工作线程都有自己的工作队列，这是使用双端队列（或者叫做 deque）来实现的（Java 6 在类库中添加了几种 deque 实现，包括 ArrayDeque 和 LinkedBlockingDeque）。当一个任务划分一个新线程时，它将自己推到 deque 的头部。当一个任务执行与另一个未完成任务的合并操作时，它会将另一个任务推到队列头部并执行，而不会休眠以等待另一任务完成（像 Thread.join() 的操作一样）。当线程的任务队列为空，它将尝试从另一个线程的 deque 的尾部 窃取另一个任务。  
可以使用标准队列实现工作窃取，但是与标准队列相比，deque 具有两方面的优势：减少争用和窃取。因为只有工作线程会访问自身的 deque 的头部，deque 头部永远不会发生争用；因为只有当一个线程空闲时才会访问 deque 的尾部，所以也很少存在线程的 deque 尾部的争用（在 fork-join 框架中结合 deque 实现会使这些访问模式进一步减少协调成本）。跟传统的基于线程池的方法相比，减少争用会大大降低同步成本。此外，这种方法暗含的后进先出（last-in-first-out，LIFO）任务排队机制意味着最大的任务排在队列的尾部，当另一个线程需要窃取任务时，它将得到一个能够分解成多个小任务的任务，从而避免了在未来窃取任务。因此，工作窃取实现了合理的负载平衡，无需进行协调并且将同步成本降到了最小。  
  
![点击查看原始大小图片][d4abd3fb-5ed0-3609-834c-96f7b89c1515.jpg]  
  
![点击查看原始大小图片][a86966e0-6fb0-3606-b84f-67a086eb2bb7.jpg]  
  
**标准实现**  
除了能够使用fork/join框架来实现能够在多处理系统中被并行执行的定制化算法（如前文中的ForkBlur.java例子），在Java SE中一些比较常用的功能点也已经使用fork/join框架来实现了。在Java SE 8中，java.util.Arrays类的一系列parallelSort()方法就使用了fork/join来实现。这些方法与sort()系列方法很类似，但是通过使用fork/join框架，借助了并发来完成相关工作。在多处理器系统中，对大数组的并行排序会比串行排序更快。这些方法究竟是如何运用fork/join框架并不在本教程的讨论范围内。想要了解更多的信息，请参见Java API文档。 其他采用了fork/join框架的方法还包括java.util.streams包中的一些方法，此包是作为Java SE 8发行版中Project Lambda的一部分。想要了解更多信息，请参见Lambda Expressions一节。

[icon_copy.gif]: http://phl.iteye.com/images/icon_copy.gif
[_icon_copy.gif]: http://phl.iteye.com/blog/2036899#
[icon_star.png]: http://phl.iteye.com/images/icon_star.png
[spinner.gif]: /images/20220810/97ad344f7c184cbebacd18e5ee7ace21.png
[d4abd3fb-5ed0-3609-834c-96f7b89c1515.jpg]: /images/20220810/e31441d667334271a157e1ae19fa81cf.png
[a86966e0-6fb0-3606-b84f-67a086eb2bb7.jpg]: /images/20220810/001c834bcb074ef99cced24e48c1badd.png