Java多线程——Fork/Join框架

使用多线程技术，让不同的任务在不同的线程中运行，可以提高CPU的利用率。然而只是仅仅通过任务来划分，粒度还是有点大。当任务中有些任务运行所占时间远远大于其它任务的时候，使用多线程并发所带来的效果并不明显。在这种情况下把耗时长的大任务划分为耗时短的小任务，可以有效地缓解并发的瓶颈，最终把小任务的结果合并获取完整的任务的结果。Fork/Join就可以很好地实现这种划分任务最后合并的思路（你可以把它当做并发版的分治）。

一、ForkJoinTask & ForkJoinPool

1. ForkJoinTask<V>

扩展自Future，它代表的是比任务更加轻量级，细粒度的并发单元。它本身是一个抽象类，这里提两个重要的抽象子类：RecursiveAction和RecursiveTask<V>，通过重写它们的compute方法可以定义任务，要注意的是前者的compute是不带返回值的，而后者的是有返回值的。

a. fork()方法允许ForkJoinTask任务异步执行，也允许一个新的ForkJoinTask从存在的ForkJoinTask中被启动。变种: invoke(), invokeAll()

b. 反过来, join()方法允许一个ForkJoinTask等待另一个ForkJoinTask执行完成。变种: get()

2. ForkJoinPool

继承自AbstractExecutorService，用于调度ForkJoinTask。

3. 调度机制

一个主的ForkJoinTask被提交到ForkJoinPool之后开始执行。然后ForkJoinTask开始（通过fork()等方法）启动它的子任务，并（通过get()等方法）等待它们完成。

二、使用Fork/Join框架

下面是最经典的Fork/Join框架的使用场景：

if task is small enough

execute the task

else

divide it into several smaller tasks

invoke all tasks

group results

例子：

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;

public class FJCount extends RecursiveTask {
 private String document[];
 private int start, end;
 private String word;

 FJCount (String _d[], int _s, int _e, String _w) {
  document = _d;
  start = _s;
  end = _e;
  word = _w;
 }
 @Override
 protected Object compute() {
  int result = 0;
  if (end - start < 10) {
   result = count(document, start, end, word);
  } else {
   int mid = (start + end) / 2;
   FJCount task1 = new FJCount(document, start, mid, word);
   FJCount task2 = new FJCount(document, mid+1, end, word);
   invokeAll(task1, task2);
  
   try {
    result = (int)task1.get() + (int)task2.get();
   } catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
   }
  }
  
  return result;
 }
 
 int count(String []document, int start, int end, String word) {
  int counter = 0;
  
  for (int i = start; i <= end; ++i) {
   if (document[i] == word) {
    ++counter;
   }
  }
  
  return counter;
 }

 public static void main(String ...arg) {
  String document[] = new String[100];
  
  for (int i = 0; i < 100; ++i) {
   if (i%10 == 0)
    document[i] = "Test";
   else
    document[i] = " ";
  }
  
  FJCount test = new FJCount(document, 0, 99,"Test");
  ForkJoinPool pool = new ForkJoinPool();
  pool.execute(test);
  
  do {
   System.out.println("****************************");
   System.out.printf("Main: Paralleism: %d\n", pool.getParallelism());
   System.out.printf("Main: Active Threads: %d\n", pool.getActiveThreadCount());
   System.out.printf("Main: Task Count: %d\n", pool.getQueuedTaskCount());
   System.out.printf("Main: Steal Count: %d\n", pool.getStealCount());
   
   try {
    TimeUnit.SECONDS.sleep(1);
   }catch (InterruptedException e){
    e.printStackTrace();
   }
  } while (!test.isDone());
  
  pool.shutdown();
  
  try {
   pool.awaitTermination(1,  TimeUnit.DAYS);
  }catch (InterruptedException e) {
   e.printStackTrace();
  }
  
  try {
   System.out.printf("Main: The word appears %d times in the document", test.get());
  } catch (InterruptedException | ExecutionException e) {
   e.printStackTrace();
  }  
 }
}

三、工作窃取

ForkJoinPool维护每个线程的任务列表，当某个任务完成时，它可以把挂在满负荷线程上的任务重新安排到空闲线程上去。就好比两个人搬砖，先搬完的回去帮另外那个人搬剩下的砖。这就是所谓的工作窃取，它解决了不同大小的任务之间的调度问题。工作窃取特性是ForkJoinPool和一些其它的ExecutorService的区别之一。

四、总结

Fork/Join通过把任务分割成更细粒度的单元，可以有效地减少瓶颈，提高CPU的使用率。当问题可以被分成很多可以并行处理的子问题的时候可以考虑使用Fork/Join。比如说Arrays的parallelSort方法就是通过Fork/Join来实现的。

Reference:

http://docs.oracle.com/javase/tutorial/essential/concurrency/forkjoin.html

http://www.oracle.com/technetwork/articles/java/fork-join-422606.html

http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ForkJoinTask.html

http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ForkJoinPool.html

《Java 7 Concurrency Cookbook》

《The Well-Grounded Java Developer》

Java多线程——Fork/Join框架

浏览过的版块