並發編程之fork/join(分而治之)

1.什么是分而治之

分而治之就是將一個大任務層層拆分成一個個的小任務，直到不可拆分，拆分依據定義的閾值划分任務規模。

fork/join通過fork將大任務拆分成小任務，在將小任務的結果join匯總

2.fork/join標准范式

先上圖

在使用fork/join做任務分配之前，首先得了解其中的幾個類：

ForkJoinPool：充當fork/join框架里面的管理者，最原始的任務都要交給它才能處理。它負責控制整個fork/join有多少個workerThread，workerThread的創建，激活都是由它來掌控。它還負責workQueue隊列的創建和分配，每當創建一個workerThread，它負責分配相應的workQueue。然后它把接到的活都交給workerThread去處理，它可以說是整個frok/join的容器。

        ForkJoinWorkerThread：fork/join里面真正干活的"工人"，本質是一個線程。里面有一個ForkJoinPool.WorkQueue的隊列存放着它要干的活，接活之前它要向ForkJoinPool注冊(registerWorker)，拿到相應的workQueue。然后就從workQueue里面拿任務出來處理。它是依附於ForkJoinPool而存活，如果ForkJoinPool的銷毀了,它也會跟着結束。

 ForkJoinPool.WorkQueue: 雙端隊列就是它，它負責存儲接收的任務。

ForkJoinTask：代表fork/join里面任務類型，我們一般用它的兩個子類RecursiveTask、RecursiveAction。這兩個區別在於RecursiveTask任務是有返回值，RecursiveAction沒有返回值。任務的處理邏輯包括任務的切分都集中在compute()方法里面。

 

3.廢話不多說，代碼走起

  fork/join在平時的使用過程中，一般分為同步調用和異步調用，下面是兩種情況的實例：

    /**同步用法*/
 1  public class SumArray {
    private static class SumTask extends RecursiveTask<Integer>{

        private final static int THRESHOLD = MakeArray.ARRAY_LENGTH/10;
        private int[] src; //表示我們要實際統計的數組
        private int fromIndex;//開始統計的下標
        private int toIndex;//統計到哪里結束的下標

        public SumTask(int[] src, int fromIndex, int toIndex) {
            this.src = src;
            this.fromIndex = fromIndex;
            this.toIndex = toIndex;
        }

        @Override
        protected Integer compute() {
            if(toIndex-fromIndex < THRESHOLD) {
                int count = 0;
                for(int i=fromIndex;i<=toIndex;i++) {
                    //SleepTools.ms(1);
                    count = count + src[i];
                }
                return count;
            }else {
                //fromIndex....mid....toIndex
                //1...................70....100
                int mid = (fromIndex+toIndex)/2;
                   //將任務一分為二
                SumTask left = new SumTask(src,fromIndex,mid);
                SumTask right = new SumTask(src,mid+1,toIndex);
                invokeAll(left,right);   //提交任務
                return left.join()+right.join();
            }
        }
    }

    public static void main(String[] args) {

        ForkJoinPool pool = new ForkJoinPool();
        int[] src = MakeArray.makeArray();

        SumTask innerFind = new SumTask(src,0,src.length-1);

        long start = System.currentTimeMillis();

        int a = pool.invoke(innerFind);//同步調用
        System.out.println("Task is Running.....");
        System.out.println("The count is "+innerFind.join()
                +" spend time:"+(System.currentTimeMillis()-start)+"ms"+a);
    }
}

 

/**
 *異步用法
 *
 *類說明：遍歷指定目錄（含子目錄）找尋指定類型文件
 */
public class FindDirsFiles extends RecursiveAction{

    private File path;//當前任務需要搜尋的目錄

    public FindDirsFiles(File path) {
        this.path = path;
    }

    public static void main(String [] args){
        try {
            // 用一個 ForkJoinPool 實例調度總任務
            ForkJoinPool pool = new ForkJoinPool();
            FindDirsFiles task = new FindDirsFiles(new File("F:/"));

            pool.execute(task);//異步調用

            System.out.println("Task is Running......");
            Thread.sleep(1);
            int otherWork = 0;
            for(int i=0;i<100;i++){
                otherWork = otherWork+i;
            }
            System.out.println("Main Thread done sth......,otherWork="+otherWork);
            task.join();//阻塞的方法
            System.out.println("Task end");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    protected void compute() {
        
        List<FindDirsFiles> subTasks = new ArrayList<>();
        
        File[] files = path.listFiles();
        if(files!=null) {
            for(File file:files) {
                if(file.isDirectory()) {
                    subTasks.add(new FindDirsFiles(file));
                }else {
                    //遇到文件，檢查
                    if(file.getAbsolutePath().endsWith("txt")) {
                        System.out.println("文件："+file.getAbsolutePath());
                    }
                }
            }
            if(!subTasks.isEmpty()) {
                for(FindDirsFiles subTask:invokeAll(subTasks)) {
                    subTask.join();//等待子任務執行完成
                }
            }
        }


        
    }
}

 

 這段代碼可以直接運行試試，跟上面的標准范式一樣，在這里我是實現了RecursiveTask(有興趣的可以自己改成RecursiveAction玩玩，但是RecursiveAction是沒有返回值的，使用的時候需要注意)，當調用ForkJoinPool的invoke方法啟動任務，會同步調用重寫的compute方法，這個方法里面才是你要寫的fork/join業務代碼。

可以看到，我定義了一個閾值THRESHOLD，當任務小於這個閾值的時候，執行運算，否則繼續切分任務，提交任務，循環調用，直到任務不可切分，將所有的運算結果整合。其實我在調用invokeAll方法時，並不會立刻返回結果，里面還是會去重復判斷每一個任務是否小於閾值，當所有的任務都滿足條件並執行完成，才會返回，其實就是遞歸調用。

 

總結：

    fork/join的使用其實沒什么難度，其基本思想是將大任務分割成小任務，最后將小任務聚合起來得到結果。fork是分解的意思, join是收集的意思. 它非常類似於HADOOP提供的MapReduce框架，只是MapReduce的任務可以針對集群內的所有計算節點，可以充分利用集群的能力完成計算任務。ForkJoin更加類似於單機版的MapReduce。