并发—Fork-Join框架

Fork-Join框架：

• 是一个用于并行执行任务的框架，是一个将大任务分解成多个子任务，并且组合多个子任务的结果来获得大任务的答案的框架。
• Fork 就是把一个大任务切分为若干个子任务并行的执行，Join就是合并这些任务的执行结果，最后得到这个大任务的结果
• 主要用于完成计算密集型任务。使用了一种有效地智能方法来平衡可用线程的工作负载，这种方法称为工作密取（work stealing）

　　　　

　　　　　　　　　　　　　　　　　　图 1 ——  Fork-Join 模式示意图

　　　　

　　　　　　　　　　　　　　　　图 2 —— 工作密取示意图

　　RecursiveAction是ForkJoinTask的一个子类，代表了一类最简单的 ForkJoinTask：

　　例子：

class Counter extends RecursiveTask<Integer> { 
    //. . . 
    protected Integer compute() { 
        if (to - from < THRESHOLD) { 
            //solve problem directly 
        } else { 
            int mid = (from + to) / 2; 
            Counter first = new Counter(values, from, mid, filter);
            Counter second = new Counter(values, mid, to, filter); 
            invokeAll(first, second); return first.join() + second.join(); 
        } 
    } 
}

　　具体程序：

package forkJoin; 
import java.util.concurrent.*; 
/** 
 * This program demonstrates the fork-join framework. 
 * @version 1.00 2012-05-20 
 * @author Cay Horstmann 
 */ 
public class ForkJoinTest { 
    public static void main(String[] args) { 
        final int SIZE = 10000000; 
        double[] numbers = new double[SIZE]; 
        for (int i = 0; i < SIZE; i++) 
            numbers[i] = Math.random(); 
        Counter counter = new Counter(numbers, 0, numbers.length, 
           new Filter() { 
                public boolean accept(double x) { return x > 0.5; } 
        }); 
        ForkJoinPool pool = new ForkJoinPool(); 
        pool.invoke(counter);
        System.out.println(counter.join()); 
    } 
} 

interface Filter { 
    boolean accept(double t); 
} 

class Counter extends RecursiveTask<Integer> { 
    public static final int THRESHOLD = 1000; 
    private double[] values; 
    private int from; 
    private int to; 
    private Filter filter;
     
    public Counter(double[] values, int from, int to, Filter filter) { 
        this.values = values; 
        this.from = from; 
        this.to = to; 
        this.filter = filter; 
    }
     
    protected Integer compute() { 
        if (to - from < THRESHOLD) { 
            int count = 0; 
            for (int i = from; i < to; i++) { 
                if (filter.accept(values[i])) 
                    count++; 
            } 
            return count; 
        } else { 
            int mid = (from + to) / 2; 
            Counter first = new Counter(values, from, mid, filter); 
            Counter second = new Counter(values, mid, to, filter); 
            invokeAll(first, second); 
            return first.join() + second.join(); 
        } 
    } 
}