輕輕松松吃透Java並發fork/join框架

 class Fibonacci extends RecursiveTask<Integer> {
   final int n;
   Fibonacci(int n) { this.n = n; }
   Integer 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();
   }
 }

// Node 1 : 返回結果 , 存放最終結果
V result;
​
// Node 2 : 抽象方法 compute , 用於計算最終結果
protected abstract V compute();
​
// Node 3 : 獲取最終結果
public final V getRawResult() {
        return result;
}
​
// Node 4 : 最終執行方法 , 這裡是需要調用具體實現類compute
protected final boolean exec() {
    result = compute();
    return true;
}

@ 
public class ForkJoinPoolService extends RecursiveTask<Integer> {
​
    private static final int THRESHOLD = 2; //閥值
    private int start;
    private int end;
​
    public ForkJoinPoolService(Integer start, Integer end) {
        this.start = start;
        this.end = end;
    }
​
    @Override
    protected Integer compute() {
        int sum = 0;
        boolean canCompute = (end - start) <= THRESHOLD;
        if (canCompute) {
            for (int i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            int middle = (start + end) / 2;
            ForkJoinPoolService leftTask = new ForkJoinPoolService(start, middle);
            ForkJoinPoolService rightTask = new ForkJoinPoolService(middle + 1, end);
            //執行子任務
            leftTask.fork();
            rightTask.fork();
            //等待子任務執行完，並得到其結果
            Integer rightResult = rightTask.join();
            Integer leftResult = leftTask.join();
            //合並子任務
            sum = leftResult + rightResult;
        }
        return sum;
    }
​
}

/**
 * 這是一個簡單的Join/Fork計算過程，將1—1001數字相加
 */
public class TestForkJoinPool {

    private static final Integer MAX = 200;

    static class MyForkJoinTask extends RecursiveTask<Integer> {
        // 子任務開始計算的值
        private Integer startValue;

        // 子任務結束計算的值
        private Integer endValue;

        public MyForkJoinTask(Integer startValue , Integer endValue) {
            this.startValue = startValue;
            this.endValue = endValue;
        }

        @Override
        protected Integer compute() {
            // 如果條件成立，說明這個任務所需要計算的數值分為足夠小瞭
            // 可以正式進行累加計算瞭
            if(endValue - startValue < MAX) {
                System.out.println("開始計算的部分：startValue = " + startValue + ";endValue = " + endValue);
                Integer totalValue = 0;
                for(int index = this.startValue ; index <= this.endValue  ; index++) {
                    totalValue += index;
                }
                return totalValue;
            }
            // 否則再進行任務拆分，拆分成兩個任務
            else {
                MyForkJoinTask subTask1 = new MyForkJoinTask(startValue, (startValue + endValue) / 2);
                subTask1.fork();
                MyForkJoinTask subTask2 = new MyForkJoinTask((startValue + endValue) / 2 + 1 , endValue);
                subTask2.fork();
                return subTask1.join() + subTask2.join();
            }
        }
    }

    public static void main(String[] args) {
        // 這是Fork/Join框架的線程池
        ForkJoinPool pool = new ForkJoinPool();
        ForkJoinTask<Integer> taskFuture =  pool.submit(new MyForkJoinTask(1,1001));
        try {
            Integer result = taskFuture.get();
            System.out.println("result = " + result);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace(System.out);
        }
    }
}

public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode)

......
asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
......

......
private ForkJoinPool(int parallelism,
                         ForkJoinWorkerThreadFactory factory,
                         UncaughtExceptionHandler handler,
                         int mode,
                         String workerNamePrefix) {
        this.workerNamePrefix = workerNamePrefix;
        this.factory = factory;
        this.ueh = handler;
        this.config = (parallelism & SMASK) | mode;
        long np = (long)(-parallelism); // offset ctl counts
        this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
    }
......

package test.thread.pool.merge;

import java.util.Arrays;
import java.util.Random;

/**
 * 歸並排序
 * @author yinwenjie
 */
public class Merge1 {

    private static int MAX = 10000;

    private static int inits[] = new int[MAX];

    // 這是為瞭生成一個數量為MAX的隨機整數集合，準備計算數據
    // 和算法本身並沒有什麼關系
    static {
        Random r = new Random();
        for(int index = 1 ; index <= MAX ; index++) {
            inits[index - 1] = r.nextInt(10000000);
        }
    }

    public static void main(String[] args) {
        long beginTime = System.currentTimeMillis();
        int results[] = forkits(inits); 
        long endTime = System.currentTimeMillis();
        // 如果參與排序的數據非常龐大，記得把這種打印方式去掉
        System.out.println("耗時=" + (endTime - beginTime) + " | " + Arrays.toString(results));       
    }

    // 拆分成較小的元素或者進行足夠小的元素集合的排序
    private static int[] forkits(int source[]) {
        int sourceLen = source.length;
        if(sourceLen > 2) {
            int midIndex = sourceLen / 2;
            int result1[] = forkits(Arrays.copyOf(source, midIndex));
            int result2[] = forkits(Arrays.copyOfRange(source, midIndex , sourceLen));
            // 將兩個有序的數組，合並成一個有序的數組
            int mer[] = joinInts(result1 , result2);
            return mer;
        } 
        // 否則說明集合中隻有一個或者兩個元素，可以進行這兩個元素的比較排序瞭
        else {
            // 如果條件成立，說明數組中隻有一個元素，或者是數組中的元素都已經排列好位置瞭
            if(sourceLen == 1
                || source[0] <= source[1]) {
                return source;
            } else {
                int targetp[] = new int[sourceLen];
                targetp[0] = source[1];
                targetp[1] = source[0];
                return targetp;
            }
        }
    }

    /**
     * 這個方法用於合並兩個有序集合
     * @param array1
     * @param array2
     */
    private static int[] joinInts(int array1[] , int array2[]) {
        int destInts[] = new int[array1.length + array2.length];
        int array1Len = array1.length;
        int array2Len = array2.length;
        int destLen = destInts.length;

        // 隻需要以新的集合destInts的長度為標準，遍歷一次即可
        for(int index = 0 , array1Index = 0 , array2Index = 0 ; index < destLen ; index++) {
            int value1 = array1Index >= array1Len?Integer.MAX_VALUE:array1[array1Index];
            int value2 = array2Index >= array2Len?Integer.MAX_VALUE:array2[array2Index];
            // 如果條件成立，說明應該取數組array1中的值
            if(value1 < value2) {
                array1Index++;
                destInts[index] = value1;
            }
            // 否則取數組array2中的值
            else {
                array2Index++;
                destInts[index] = value2;
            }
        }

        return destInts;
    }
}

......
/**
 * 使用Fork/Join框架的歸並排序算法
 * @author yinwenjie
 */
public class Merge2 {

    private static int MAX = 100000000;

    private static int inits[] = new int[MAX];

    // 同樣進行隨機隊列初始化，這裡就不再贅述瞭
    static {
        ......
    }

    public static void main(String[] args) throws Exception {   
        // 正式開始
        long beginTime = System.currentTimeMillis();
        ForkJoinPool pool = new ForkJoinPool();
        MyTask task = new MyTask(inits);
        ForkJoinTask<int[]> taskResult = pool.submit(task);
        try {
            taskResult.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace(System.out);
        }
        long endTime = System.currentTimeMillis();
        System.out.println("耗時=" + (endTime - beginTime));      
    }

    /**
     * 單個排序的子任務
     * @author yinwenjie
     */
    static class MyTask extends RecursiveTask<int[]> {

        private int source[];

        public MyTask(int source[]) {
            this.source = source;
        }

        /* (non-Javadoc)
         * @see java.util.concurrent.RecursiveTask#compute()
         */
        @Override
        protected int[] compute() {
            int sourceLen = source.length;
            // 如果條件成立，說明任務中要進行排序的集合還不夠小
            if(sourceLen > 2) {
                int midIndex = sourceLen / 2;
                // 拆分成兩個子任務
                MyTask task1 = new MyTask(Arrays.copyOf(source, midIndex));
                task1.fork();
                MyTask task2 = new MyTask(Arrays.copyOfRange(source, midIndex , sourceLen));
                task2.fork();
                // 將兩個有序的數組，合並成一個有序的數組
                int result1[] = task1.join();
                int result2[] = task2.join();
                int mer[] = joinInts(result1 , result2);
                return mer;
            } 
            // 否則說明集合中隻有一個或者兩個元素，可以進行這兩個元素的比較排序瞭
            else {
                // 如果條件成立，說明數組中隻有一個元素，或者是數組中的元素都已經排列好位置瞭
                if(sourceLen == 1
                    || source[0] <= source[1]) {
                    return source;
                } else {
                    int targetp[] = new int[sourceLen];
                    targetp[0] = source[1];
                    targetp[1] = source[0];
                    return targetp;
                }
            }
        }

        private int[] joinInts(int array1[] , int array2[]) {
            // 和上文中出現的代碼一致
        }
    }
}