基於Fork/Join框架實現對大型浮點數數組排序（歸併算法和插入排序算法）

分支/合併框架

說明

重點是那個浮點數數組排序的例子，從主函數展開，根據序號看

一、GitHub代碼歡迎star。大家輕輕的一點，對我鼓勵特大，我有一個習慣，看完別人的文章是會點讚的。

二、我的認爲學習語言最好的方式就是模仿、思考別人爲何這麼寫。結合栗子效果更好，也能記住知識點。

三、只由於本身知識欠缺，語言組織能力不行，因此只能以這樣方式記錄。感受效果很好。

四、過年前一天，過年中、過年第一天。聽力三遍《咱們不同》，感謝大家的鼓勵，點贊。

前面的知識點都利用率 如今計算機提供的並行機制，然而咱們還須要粒度更細的並行機制。例如：考慮使用遞歸計算Fibonacci數列的方法，

finonacci( n - 1) + finonacci( n - 2)

能夠將這兩個子任務分配給每一個新的線程，當他們計算完成時，將結果相加。事實上，每一個字問題的計算又能夠分解爲兩個子問題，直到不可細分位置 這類算法被稱爲分治算法複雜的問題被分解爲較小的問題，在根據字問題的解推導出原始問題的解。這樣問題就容易並行化。Java SE 7引入了新的分支/合併(Fork/Join)框架以簡化這類分治算法的實現。

大型任務被分解爲若干塊，而後放入隊列用於後續計算，在隊列中，任務還能夠將自身分解爲更小的部分。線程會從隊列中可以如取出任務並執行。當全部線程結束後 ，將各部分結果合併獲得最終結果。「分支」是指任務分解，「合併」是指結果合併。每一個工做線程都維着任務的雙端隊列。隊列中後來的任務先執行。當工做中沒有任務時須要執行的時候，會嘗試從其餘線程「竊取」任務，若是」竊取失敗，沒有其餘 工做可作，就會下線。「竊取」的好處是減小了工做隊列中爭用狀況。

像Task這樣的大型任務將被分解爲兩個或更多的子任務，每一個子任務能夠進一步分解爲新的子任務。直到子任務變得足夠簡單並獲得解決。子任務敵對獲得解決。

ForkJoinPool類

ForkJoinPool類是用於執行ForkJoinTask的ExecutorSerivce。與其餘ExecutorService的不一樣之處在於ForkJoinPool採用了前面提到的「工做竊取」機制。在構造過程當中，能夠在構造函中指定線程池的大小。若是使用的是默認無參構造函數，那麼會建立大小等於可用處理器數量的線程池。儘管已之地功能線程池的大小，但線程還會在嘗試維護更多活躍線程的任意時刻動態調整自身大小。ForkJoinPool提供了相應方法用於管理和監控那些提交的任務。ForkJoinPool與其餘ExecutorService的另外一個重大區別在於：線程池須要在程序結束時顯示中止,由於其中全部的線程都處於守護狀態

有三種不一樣的方式能夠將任務提交給ForkJoinPool。在異步執行模式下，能夠調用execute方法，並將ForkJoinTask做爲參數。至於任務自己須要調用fork方法將任務在多個線程間分解。若是須要等待計算結果，能夠調用ForkJoinPool的invoke方法。在ForkJoinTask中，能夠接着調用invoke方法。invoke方法開始執行任務並在任務結束後返回結果。若是底層失敗就會拋出異常或錯誤。最後能夠經過調用ForkJoinPool的submit方法將任務提交給線程池，submit會返回一個Future對象，可使用該對象檢查任務狀態和獲取執行任務的結果。

ForkJoinTask類

ForkJoinTask類是運行在前面提到的ForkJoinPool中，用來建立任務的抽象類，RecursiveAction和RecursiveTask僅有兩個直接子類。任務在作提交給ForkJoinPool後，便開始執行。ForkJoinTask僅包含兩個操做---分支和合並一旦開始執行，就會啓動其餘子任務。合併操做會等待子操做結束並在結果後提取運行結果。ForkJoinTask實現類Future接口，是Fiture輕量級形式。Future接口的get方法實現，可用於等待任務結束並取得結果。還能夠經過invoke方法執行任務。該方法會在任務結束後返回結果，invokeAl方法能夠接受任務集合做爲參數，該方法會分解出指定集合中的全部任務，並在每一個人物結束後返回，或異常時。

ForkJoinTask類提供若干檢查任務執行狀態的方法，只要任務結束，無論什麼方法，isDone方法都用於檢查任務是否結束，返回true表示任務結束。isCompleledAbnormally方法用於檢測任務是否在被取消而且沒有異常的狀況下正常結束。返回true表示正常結束。
isCancelled方法用於檢查任務是否被取消，返回true表示正常取消。

一般狀況下不會直接繼承ForkJoinTask類，相反，會建立基於RecursiveTask或RecursiveAction的類。二者均爲ForkJoinTask的抽象子類。RucursiveTask用於返回結果的任務， 而RecursiveAction用於不返回結果的任務。不管使用哪種，都要在子類中實現compute方法，並在其中執行任務的主要計算。

大型浮點數數組排序

/**
 * Created by guo on 16/2/2018.
 * 大型浮點數數組排序
 * 需求：
 *      一、假設有一些數目很是大的浮點數(一百萬個)，須要編寫一個程序將這些數字按升序排序，
 *      二、針對單線程的經常使用排序算法須要消耗過長的時間才能生成排好序的數組。
 *      三、這類問題剛好與分治模式相契合。咱們將數組分解成多個較小的數組，並在每一個數組中獨立進行排序。
 *      四、最後將不斷歸併排好序的數組合併成更大的數組以建立最終排好序的數組。
 */
public class ParallelMergeSort {
    private static ForkJoinPool threadPool;
    private static final int THRESHOLD = 16;

    /**
     * 四、1 sort方法接受對象數組做爲參數，目標是對數組進行升序排序。
     * @param objectArray
     */
    private static void sort(Comparable[] objectArray) {
        //四、2 聲明臨時目標數組用於存儲排序結果，大小等同於輸入數組。
        Comparable[] destArray = new Comparable[objectArray.length];
        //四、3 建立一個SortTask對象，並調用invoke方法將它提交給線程池。
        //四、4 SortTask接受4個參數，待排序的數組，已經用於存儲排序後的對象目標數組，源數組中待排序元素的開始索引與結束索引。
        threadPool.invoke(new SortTask(objectArray, destArray, 0, objectArray.length - 1));
    }

    /**
     * 重點：
     *    --五、SortTask其功能繼承自RecursiveAction類。
     *      a、此排序算法不直接返回結果給調用方，所以基於RecursiveAction類。
     *      b、若是算法具備返回值，好比計算 finonacci( n - 1) + finonacci( n - 2) ，那麼就應該繼承RecursiveTask類哦。
     *      c、做爲SortTask類的具體實現的一部分，須要重寫compute抽象方法，
     */
    static class SortTask extends RecursiveAction {
        private Comparable[] sourceArray;
        private Comparable[] destArray;
        private int lowerIndex, upperIndex;

        public SortTask(Comparable[] sourceArray, Comparable[] destArray, int lowerIndex, int upperIndex) {
            this.sourceArray = sourceArray;
            this.destArray = destArray;
            this.lowerIndex = lowerIndex;
            this.upperIndex = upperIndex;
        }

        /**
         * 六、compute方法用於檢查帶排序原色的大小。
         */
        @Override
        protected void compute() {
              //六、1 若是小於預約義的值THRESHOLD(16),就調用insertSort方法進行排序 。
            if (upperIndex - lowerIndex < THRESHOLD) {
                insertionSort(sourceArray, lowerIndex, upperIndex);
                return;
            }

            //六、2 若是沒有超過，就建立兩個子任務並遞歸進行調用。
            //六、3 每一個子任務接收原有數組的一半做爲本身的源數組，
            //六、3 minIndex定義了原有數組的中心點。調用invokeAll，將這兩我的物提交給線程池
            //六、4 分解任務爲子任務的過程會一直遞歸執行，直到每一個子任務變得足夠小位置。
            //六、5 全部分解獲得的子任務都被遞交給線程池，當它們結束時，compute方法會調用merge方法
            int minIndex = (lowerIndex + upperIndex >>> 1);
            invokeAll(new SortTask(sourceArray, destArray, lowerIndex, minIndex), new SortTask(sourceArray, destArray, minIndex + 1, upperIndex));
            merge(sourceArray, destArray, lowerIndex, minIndex, upperIndex);
        }
    }

    /**
     *歸併算法
     */
    public static void merge(Comparable[] sourceArray, Comparable[] destArray, int lowerIndex, int mindIndex, int upperIndex) {
        //一、 若是源數組中間索引小於等於右邊的則直接返回。
        if (sourceArray[mindIndex].compareTo(sourceArray[mindIndex + 1]) <= 0) {
            return;
        }
        /**
         * 二、調用底層實現的數組拷貝，是一個本地方法
         * void arraycopy(Object src,  int  srcPos,Object dest, int destPos,int length);
         *  src      the source array.
         * srcPos   starting position in the source array.
         * dest     the destination array.
         * destPos  starting position in the destination data.
         * length   the number of array elements to be copied.
         */
        System.arraycopy(sourceArray, lowerIndex, destArray, lowerIndex, mindIndex - lowerIndex + 1);
        int i = lowerIndex;
        int j = mindIndex + 1;
        int k = lowerIndex;
         //三、 將兩個數組進行歸併，
        while (k < j && j < upperIndex) {
            if (destArray[i].compareTo(sourceArray[j]) <= 0) {
                sourceArray[k++] = destArray[i++];
            } else {
                sourceArray[k++] = sourceArray[j++];
            }
        }
        System.arraycopy(destArray, i, sourceArray, k, j - k);
    }

    /**
     * 插入排序 (得好好研究下算法了)
     * 一、從後向前找到格式的位置插入，
     * 二、 每步將一個待排序的記錄，按其順序大小插入到前面已經排好的子序列合適的位置。
     * 三、直到所有插入位置。
     */
    private static void insertionSort(Comparable[] objectArray, int lowerIndex, int upperIndex) {
        //一、控制比較的輪數，
        for (int i = lowerIndex + 1; i <= upperIndex; i++) {
            int j = i;
            Comparable tempObject = objectArray[j];
            //二、後一個和前面一個比較，若是前面的小，則把前面的賦值到後面。
            while (j > lowerIndex && tempObject.compareTo(objectArray[j - 1]) < 0) {
                objectArray[j] = objectArray[j - 1];
                --j;
            }
            objectArray[j] = tempObject;
        }
    }

    /**
     * 三、1 使用Random類生成範圍在0-1000之間的數據點，並使用這些隨機生成的數據初始化數組每個元素。
     * 三、2 建立好的數據點的數目等同於函數參數的數目，這個例子是1000，增大該數字，能夠驗證並行排序算法的效率。
     * @param length
     * @return
     */
    public static Double[] createRandomData(int length) {
        Double[] data = new Double[length];
        for (int i = 0; i < data.length; i++) {
            data[i] = length * Math.random();
        }
        return data;
    }
}

主函數

/**
     * 主函數
     *
     * @param args
     */
    public static void main(String[] args) {
        //一、獲取當前運行代碼所在機器上的能夠用處理器數目
        int processors = Runtime.getRuntime().availableProcessors();
        System.out.println("no of processors:" + processors);
        //二、1建立大小等同於處理器數目的線程池，這一數目是運行在可用硬件最佳數目
        //二、2若是建立更大的線程池，就會有CPU競爭的狀況發生，
        //二、3經過實例化ForkJoinPool，並將線程池大小做爲參數傳入構造函數以建立線程池
        threadPool = new ForkJoinPool(processors);
        //三、構造隨機數據的數組，並輸入以方便嚴重
        Double[] data = createRandomData(100000);
        System.out.println("original unsorted data：");
        for (Double d : data) {
            System.out.printf("%3.2f ", (double) d);
        }
        //四、調用sort方法對生成的數據進行排序，並再次輸出數組，以驗證數組已經排好序。
        sort(data);

        System.out.println("\n\n Sorted Array ");
        for (double d : data) {
            System.out.printf("%3.2f ", d);
        }
    }