併發—Fork-Join框架

Fork-Join框架：

• 是一個用於並行執行任務的框架，是一個將大任務分解成多個子任務，而且組合多個子任務的結果來得到大任務的答案的框架。
• Fork 就是把一個大任務切分爲若干個子任務並行的執行，Join就是合併這些任務的執行結果，最後獲得這個大任務的結果
• 主要用於完成計算密集型任務。使用了一種有效地智能方法來平衡可用線程的工做負載，這種方法稱爲工做密取（work stealing）

　　　　

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

　　　　

　　　　　　　　　　　　　　　　圖 2 —— 工做密取示意圖

　　RecursiveAction是ForkJoinTask的一個子類，表明了一類最簡單的 ForkJoinTask：

　　例子：

 1 class Counter extends RecursiveTask<Integer> { 
 2     //. . . 
 3     protected Integer compute() { 
 4         if (to - from < THRESHOLD) { 
 5             //solve problem directly 
 6         } else { 
 7             int mid = (from + to) / 2; 
 8             Counter first = new Counter(values, from, mid, filter);
 9             Counter second = new Counter(values, mid, to, filter); 
10             invokeAll(first, second); return first.join() + second.join(); 
11         } 
12     } 
13 }

　　具體程序：

 1 package forkJoin; 
 2 import java.util.concurrent.*; 
 3 /** 
 4  * This program demonstrates the fork-join framework. 
 5  * @version 1.00 2012-05-20 
 6  * @author Cay Horstmann 
 7  */ 
 8 public class ForkJoinTest { 
 9     public static void main(String[] args) { 
10         final int SIZE = 10000000; 
11         double[] numbers = new double[SIZE]; 
12         for (int i = 0; i < SIZE; i++) 
13             numbers[i] = Math.random(); 
14         Counter counter = new Counter(numbers, 0, numbers.length, 
15            new Filter() { 
16                 public boolean accept(double x) { return x > 0.5; } 
17         }); 
18         ForkJoinPool pool = new ForkJoinPool(); 
19         pool.invoke(counter);
20         System.out.println(counter.join()); 
21     } 
22 } 
23 
24 interface Filter { 
25     boolean accept(double t); 
26 } 
27 
28 class Counter extends RecursiveTask<Integer> { 
29     public static final int THRESHOLD = 1000; 
30     private double[] values; 
31     private int from; 
32     private int to; 
33     private Filter filter;
34      
35     public Counter(double[] values, int from, int to, Filter filter) { 
36         this.values = values; 
37         this.from = from; 
38         this.to = to; 
39         this.filter = filter; 
40     }
41      
42     protected Integer compute() { 
43         if (to - from < THRESHOLD) { 
44             int count = 0; 
45             for (int i = from; i < to; i++) { 
46                 if (filter.accept(values[i])) 
47                     count++; 
48             } 
49             return count; 
50         } else { 
51             int mid = (from + to) / 2; 
52             Counter first = new Counter(values, from, mid, filter); 
53             Counter second = new Counter(values, mid, to, filter); 
54             invokeAll(first, second); 
55             return first.join() + second.join(); 
56         } 
57     } 
58 }