JDK7新特性<五> fork/join 框架

对于框架的原理，可以阅读 Doug Lea 的文章“A Java Fork/Join Framework”：了解 Fork/Join 模式的实现机制和执行性能。

 

 

原理解析：fork分解，join结合。这个框架的本质是将一个任务分解成多个子任务，每个子任务用单独的线程去处理。这里用到了递归的思想。框架的结构图可以参考

 图片来源（http://www.ibm.com/developerworks/cn/java/j-lo-forkjoin/index.html）

 

 

使用fork/join 框架很简单，

1.实现子问题的一般求解算法

2.如何分解问题

3.继承 RecursiveAction ，实现compute()方法

 

      Result solve(Problem problem) {
	if (problem is small)
		directly solve problem
	else {
		split problem into independent parts
		fork new subtasks to solve each part
		join all subtasks
		compose result from subresults
	}
}

 

 

 

这里我通过一个改进的二分查找来讲解fork/join的使用。（后面才发现，选用这个案例是非常失败的，因为二分查找的时间是logn,而创建线程的开销更大，这样并不能体现多线程二分查找的优势，所以这个代码不具有实用性，只是为了说明如何使用框架：）

 

代码如下：

BinarySearchProblem.java

 

 

package testjdk7;

import java.util.Arrays;
/**
 * @author kencs@foxmail.com
 */
public class BinarySearchProblem {
    private final int[] numbers;
    private final int start;
    private final int end;
    public final int size;
    
    public BinarySearchProblem(int[] numbers,int start,int end){
        this.numbers = numbers;
        this.start = start;
        this.end = end;
        this.size = end -start;
    }
    
    public int searchSequentially(int numberToSearch){
       //偷懒，不自己写二分查找了
       return Arrays.binarySearch(numbers, start, end, numberToSearch);
    }
    
    public BinarySearchProblem subProblem(int subStart,int subEnd){
        return new BinarySearchProblem(numbers,start+subStart,start+subEnd);
    }
}

 

 BiSearchWithForkJoin.java

 

package testjdk7;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

/**
 * @author kencs@foxmail.com
 */
public class BiSearchWithForkJoin extends RecursiveAction {
    private final int threshold;
    private final BinarySearchProblem problem;
    public int result;
    private final int numberToSearch;
    
    public BiSearchWithForkJoin(BinarySearchProblem problem,int threshold,int numberToSearch){
        this.problem = problem;
        this.threshold = threshold;
        this.numberToSearch = numberToSearch;
    }

    @Override
    protected void compute() {
       if(problem.size < threshold){ //小于阀值，就直接用普通的二分查找
           result = problem.searchSequentially(numberToSearch);
       }else{
           //分解子任务
           int midPoint = problem.size/2;
           BiSearchWithForkJoin left = new BiSearchWithForkJoin(problem.subProblem(0, midPoint),threshold,numberToSearch);
           BiSearchWithForkJoin right = new BiSearchWithForkJoin(problem.subProblem(midPoint+1, problem.size),threshold,numberToSearch);
           invokeAll(left,right);
           result = Math.max(left.result, right.result);
       }
    }
    
    //构造数据
    private static final int[] data = new int[1000_0000];
    static{
        for(int i = 0;i<1000_0000;i++){
            data[i] = i;
        }
    }
    public static void main(String[] args){
       BinarySearchProblem problem = new BinarySearchProblem(data,0,data.length);
       int threshold = 100;
       int nThreads = 10;
       //查找100_0000所在的下标
       BiSearchWithForkJoin  bswfj = new BiSearchWithForkJoin(problem,threshold,100_0000);
       ForkJoinPool fjPool = new ForkJoinPool(nThreads);
       fjPool.invoke(bswfj);
       System.out.printf("Result is:%d%n",bswfj.result);
    }
    
    
}

  RecursiveTask 还可以带返回值，这里给出一段代码作为参考（斐波那契函数）

（来自http://www.ibm.com/developerworks/cn/java/j-lo-forkjoin/index.html）

 

 

class Fibonacci extends RecursiveTask<Integer> {
    final int n;

    Fibonacci(int n) {
        this.n = n;
    }

    private int compute(int small) {
        final int[] results = { 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89 };
        return results[small];
    }

    public Integer compute() {
        if (n <= 10) {
            return compute(n);
        }
        Fibonacci f1 = new Fibonacci(n - 1);
        Fibonacci f2 = new Fibonacci(n - 2);
        System.out.println("fork new thread for " + (n - 1));
        f1.fork();
        System.out.println("fork new thread for " + (n - 2));
        f2.fork();
        return f1.join() + f2.join();
    }
}

 

 

用途

    只要问题能够分解成类似子问题的，都可以使用这个框架。对于大批量的数据尤其合适

 

  参考资料

    Jdk7官网 http://openjdk.java.net/projects/jdk7/

 

 

 

 

 

 

   （注：这篇文章发表时，JDK7未正式公布，可能有误差，具体以官方正式版为准）

 

更多的jdk7文章，欢迎访问http://janeky.iteye.com/category/157060

 

 

 

我们在实际应用中有如下应用场景或许可以考虑使用该并行框架来实现：

1、账户计息：系统需要对数百万账户进行利息计算和处理。如果采用JDK7的并行计算框架，那么任务的拆分可以按账号分段来实现，计算结果通常不需要汇总；

2、绩效指标计算：系统需要对数百万客户计算绩效指标，该指标计算完成后，又需要基于这些指标计算更上层的指标。对客户的指标计算可以按客户号分段，指标结果或许可以归并起来用于计算更上层指标；

但JDK7的并行计算框架只是给我们屏蔽了多线程并行执行的技术细节，其它比如：中间计算结果和计算状态的持久化问题、子任务的动态管理和结果轮询等问题还是需要我们自己开发应用框架来解决。

Map--Reduce!

把map reduce引入了。。

明明是map reduce，搞个屁名字

java7 给力！
java7在语言级别增加并发计算能力，值得期待

数据量大，算法稳定的，可以考虑用这个框架

说实话，Java 7的fork/join远比C# TPL丑陋（学师傅也只学了办拉子），原因是闭包被从Java 7中抽走了，本来可以象C# python一样写的很简单的代码成了大面条：

有闭包的Java代码应该是这样的：

double bestGpa = students.withFilter({Student s => (s.graduationYear == THIS_YEAR) }) .withMapping({ Student s => s.gpa }) .max();

特性越来越多，平台越来越臃肿了，希望不会出现最后各种编码风格混乱不堪的情形。希望可以变得更简洁更优雅

应该取消那些静态内部类，匿名类，拿掉垃圾的awt和含糊不清的io，重新定义java.util名称空间，这个包完全就是垃圾堆，无任何优雅可言。