一、Fork-Join

java下多線程的開發(fā)可以是我們自己啟用多線程，線程池，還可以使用forkjoin，forkjoin 可以讓我們不去了解諸如 Thread,Runnable 等相關(guān)的知識(shí)，只要遵循forkjoin 的開發(fā)模式，就可以寫出很好的多線程并發(fā)程序， forkjoin 在處理分而治之這一類問題時(shí)非常的有用。

1. 什么是 Fork-Join

Fork/Join框架是Java7提供的一個(gè)用于并行執(zhí)行任務(wù)的分治編程框架，是一個(gè)把大任務(wù)分割成若干個(gè)小任務(wù)，最終匯總每個(gè)小任務(wù)結(jié)果后得到大任務(wù)結(jié)果的框架，這種開發(fā)方法也叫分治編程。分治編程可以極大地利用CPU資源，提高任務(wù)執(zhí)行的效率，也是目前與多線程有關(guān)的前沿技術(shù)。

2. 分治編程會(huì)遇到什么問題

分治的原理上面說了，就是切割大任務(wù)成小任務(wù)來完成。咦，看起來好像也不難實(shí)現(xiàn)?。槭裁磳ｉT弄一個(gè)新的框架呢？我們先看一下，在不使用 Fork-Join 框架時(shí)，使用普通的線程池是怎么實(shí)現(xiàn)的。我們往一個(gè)線程池提交了一個(gè)大任務(wù)，規(guī)定好任務(wù)切割的閥值。由池中線程（假設(shè)是線程A）執(zhí)行大任務(wù)，發(fā)現(xiàn)大任務(wù)的大小大于閥值，于是切割成兩個(gè)子任務(wù)，并調(diào)用 submit() 提交到線程池，得到返回的子任務(wù)的 Future。線程A就調(diào)用 返回的 Future 的 get() 方法阻塞等待子任務(wù)的執(zhí)行結(jié)果。池中的其他線程（除線程A外，線程A被阻塞）執(zhí)行兩個(gè)子任務(wù)，然后判斷子任務(wù)的大小有沒有超過閥值，如果超過，則按照步驟2繼續(xù)切割，否則，才計(jì)算并返回結(jié)果。嘿，好像一切都很美好。真的嗎？別忘了， 每一個(gè)切割任務(wù)的線程（如線程A）都被阻塞了，直到其子任務(wù)完成，才能繼續(xù)往下運(yùn)行 。如果任務(wù)太大了，需要切割多次，那么就會(huì)有多個(gè)線程被阻塞，性能將會(huì)急速下降。更糟糕的是，如果你的線程池的線程數(shù)量是有上限的，極可能會(huì)造成池中所有線程被阻塞，線程池?zé)o法執(zhí)行任務(wù)。

@ Example1 普通線程池實(shí)現(xiàn)分治時(shí)阻塞的問題

來看一個(gè)例子，體會(huì)一下吧！下面的例子是將 1+2+...+10 的任務(wù) 分割成相加的個(gè)數(shù)不能超過3（即兩端的差不能大于2）的多個(gè)子任務(wù)。

//普通線程池下實(shí)現(xiàn)的分治效果測試
public class CommonThreadPoolTest {
    //固定大小的線程池，池中線程數(shù)量為3
    static ExecutorService fixPoolExcutor = Executors.newFixedThreadPool(3);

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        //計(jì)算 1+2+...+10  的結(jié)果
        CountTaskCallable task = new CountTaskCallable(1,10);
        //提交主人翁
        Future<Integer> future = fixPoolExcutor.submit(task);
        System.out.println("計(jì)算的結(jié)果："+future.get());
    }
}

class CountTaskCallable implements Callable<Integer> {

    //設(shè)置閥值為2
    private static final int THRESHOLD = 2;
    private int start;
    private int end;

    public CountTaskCallable(int start, int end) {
        super();
        this.start = start;
        this.end = end;
    }

    @Override
    public Integer call() throws Exception {
        int sum = 0;
        //判斷任務(wù)的大小是否超過閥值
        boolean canCompute = (end - start) <= THRESHOLD;
        if (canCompute) {
            for (int i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            System.out.println("切割的任務(wù)："+start+"加到"+end+"   執(zhí)行此任務(wù)的線程是 "+Thread.currentThread().getName());
            int middle = (start + end) / 2;
   
            CountTaskCallable leftTaskCallable = new CountTaskCallable(start, middle);
            CountTaskCallable rightTaskCallable = new CountTaskCallable(middle + 1, end);
            // 將子任務(wù)提交到線程池中
            Future<Integer> leftFuture = CommonThreadPoolTest.fixPoolExcutor.submit(leftTaskCallable);
            Future<Integer> rightFuture = CommonThreadPoolTest.fixPoolExcutor.submit(rightTaskCallable);
            //阻塞等待子任務(wù)的執(zhí)行結(jié)果
            int leftResult = leftFuture.get();
            int rightResult = rightFuture.get();
            // 合并子任務(wù)的執(zhí)行結(jié)果
            sum = leftResult + rightResult;
            
        }
        return sum;
    }

}

運(yùn)行結(jié)果

切割的任務(wù)：1加到10 執(zhí)行此任務(wù)的線程是 pool-1-thread-1
切割的任務(wù)：1加到5 執(zhí)行此任務(wù)的線程是 pool-1-thread-2
切割的任務(wù)：6加到10 執(zhí)行此任務(wù)的線程是 pool-1-thread-3

池的線程只有三個(gè)，當(dāng)任務(wù)分割了三次后，池中的線程也就都被阻塞了，無法再執(zhí)行任何任務(wù)，一直卡著動(dòng)不了。

3、Fork-Join 原理

image

4、工作竊取

針對上面的問題，F(xiàn)ork-Join 框架使用了 “工作竊?。╳ork-stealing）”算法。工作竊取（work-stealing）算法是指某個(gè)線程從其他隊(duì)列里竊取任務(wù)來執(zhí)行?？匆幌隆禞ava 并發(fā)編程的藝術(shù)》對工作竊取算法的解釋：

使用工作竊取算法有什么優(yōu)勢呢？假如我們需要做一個(gè)比較大的任務(wù)，我們可以把這個(gè)任務(wù)分割為若干互不依賴的子任務(wù)，為了減少線程間的競爭，于是把這些子任務(wù)分別放到不同的隊(duì)列里，并為每個(gè)隊(duì)列創(chuàng)建一個(gè)單獨(dú)的線程來執(zhí)行隊(duì)列里的任務(wù)，線程和隊(duì)列一一對應(yīng)，比如A線程負(fù)責(zé)處理A隊(duì)列里的任務(wù)。但是有的線程會(huì)先把自己隊(duì)列里的任務(wù)干完，而其他線程對應(yīng)的隊(duì)列里還有任務(wù)等待處理。干完活的線程與其等著，不如去幫其他線程干活，于是它就去其他線程的隊(duì)列里竊取一個(gè)任務(wù)來執(zhí)行。而在這時(shí)它們會(huì)訪問同一個(gè)隊(duì)列，所以為了減少竊取任務(wù)線程和被竊取任務(wù)線程之間的競爭，通常會(huì)使用雙端隊(duì)列，被竊取任務(wù)線程永遠(yuǎn)從雙端隊(duì)列的頭部拿任務(wù)執(zhí)行，而竊取任務(wù)的線程永遠(yuǎn)從雙端隊(duì)列的尾部拿任務(wù)執(zhí)行。

image

Fork-Join 框架使用工作竊取算法對分治編程實(shí)現(xiàn)的描述：

下面是 ForkJoin 框架對分治編程實(shí)現(xiàn)的過程的描述，增加對工作竊取算法的理解。在下面的內(nèi)容提供了一個(gè)分治的例子，可結(jié)合這部分描述一起看。

（1）Fork-Join 框架的線程池ForkJoinPool 的任務(wù)分為“外部任務(wù)” 和 “內(nèi)部任務(wù)”。
（2）“外部任務(wù)”是放在 ForkJoinPool 的全局隊(duì)列里；
（3）ForkJoinPool 池中的每個(gè)線程都維護(hù)著一個(gè)內(nèi)部隊(duì)列，用于存放“內(nèi)部任務(wù)”。
（4）線程切割任務(wù)得到的子任務(wù)就會(huì)作為“內(nèi)部任務(wù)”放到內(nèi)部隊(duì)列中。
（5）當(dāng)此線程要想要拿到子任務(wù)的計(jì)算結(jié)果時(shí)，先判斷子任務(wù)沒有完成，如果沒有完成，則再判斷子任務(wù)有沒有被其他線程“竊取”，一旦子任務(wù)被竊取了則去執(zhí)行本線程“內(nèi)部隊(duì)列”的其他任務(wù)，或者掃描其他的任務(wù)隊(duì)列，竊取任務(wù)，如果子任務(wù)沒有被竊取，則由本線程來完成。
（6）最后，當(dāng)線程完成了其“內(nèi)部任務(wù)”，處于空閑的狀態(tài)時(shí)，就會(huì)去掃描其他的任務(wù)隊(duì)列，竊取任務(wù)，盡可能地
總之，F(xiàn)orkJoin線程在等待一個(gè)任務(wù)的完成時(shí)，要么自己來完成這個(gè)任務(wù)，或者在其他線程竊取了這個(gè)任務(wù)的情況下，去執(zhí)行其他任務(wù)，是不會(huì)阻塞等待，從而避免浪費(fèi)資源，除非是所有任務(wù)隊(duì)列都為空。

工作竊取算法的優(yōu)點(diǎn)：

（1）線程是不會(huì)因?yàn)榈却硞€(gè)子任務(wù)的完成或者沒有內(nèi)部任務(wù)要執(zhí)行而被阻塞等待、掛起，而是會(huì)掃描所有的隊(duì)列，竊取任務(wù)，直到所有隊(duì)列都為空時(shí)，才會(huì)被掛起。 就如上面所說的。
（2）Fork-Join 框架在多CPU的環(huán)境下，能提供很好的并行性能。在使用普通線程池的情況下，當(dāng)CPU不再是性能瓶頸時(shí)，能并行地運(yùn)行多個(gè)線程，然而卻因?yàn)橐コ庠L問一個(gè)任務(wù)隊(duì)列而導(dǎo)致性能提高不上去。而 Fork-Join 框架為每個(gè)線程為維護(hù)著一個(gè)內(nèi)部任務(wù)隊(duì)列，以及一個(gè)全局的任務(wù)隊(duì)列，而且任務(wù)隊(duì)列都是雙向隊(duì)列，可從首尾兩端來獲取任務(wù)，極大地減少了競爭的可能性，提高并行的性能。

5、Fork/Join 實(shí)戰(zhàn)

1）Fork/Join 使用的標(biāo)準(zhǔn)范式

我們要使用 ForkJoin 框架，必須首先創(chuàng)建一個(gè) ForkJoin 任務(wù)。它提供在任務(wù)中執(zhí)行 fork 和 join 的操作機(jī)制，通常我們不直接繼承 ForkjoinTask 類，只需要直接繼承其子類。

（1）RecursiveAction，用于沒有返回結(jié)果的任務(wù)

（2） RecursiveTask，用于有返回值的任務(wù)

task 要通過 ForkJoinPool 來執(zhí)行，使用 submit 或 invoke 提交，兩者的區(qū)別是：invoke 是同步執(zhí)行，調(diào)用之后需要等待任務(wù)完成，才能執(zhí)行后面的代碼；submit 是異步執(zhí)行。

join()和 get 方法當(dāng)任務(wù)完成的時(shí)候返回計(jì)算結(jié)果。

在我們自己實(shí)現(xiàn)的 compute 方法里，首先需要判斷任務(wù)是否足夠小，如果足夠小就直接執(zhí)行任務(wù)。如果不足夠小，就必須分割成兩個(gè)子任務(wù)，每個(gè)子任務(wù)在調(diào)用 invokeAll 方法時(shí)，又會(huì)進(jìn)入 compute 方法，看看當(dāng)前子任務(wù)是否需要繼續(xù)分割成孫任務(wù)，如果不需要繼續(xù)分割，則執(zhí)行當(dāng)前子任務(wù)并返回結(jié)果。使用 join方法會(huì)等待子任務(wù)執(zhí)行完并得到其結(jié)果。

2）、Fork/Join 的同步用法和異步用法

同步示例

/**
 * forkjoin實(shí)現(xiàn)的歸并排序
 */
public class FkSort {
    private static class SumTask extends RecursiveTask<int[]>{

        private final static int THRESHOLD = 2;
        private int[] src;

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

        @Override
        protected int[] compute() {
            if(src.length<=THRESHOLD){
                return InsertionSort.sort(src);
            }else{
                //fromIndex....mid.....toIndex
                int mid = src.length / 2;
                SumTask leftTask = new SumTask(Arrays.copyOfRange(src, 0, mid));
                SumTask rightTask = new SumTask(Arrays.copyOfRange(src, mid, src.length));
                invokeAll(leftTask,rightTask);
                int[] leftResult = leftTask.join();
                int[] rightResult = rightTask.join();
                return MergeSort.merge(leftResult,rightResult);
            }
        }
    }


    public static void main(String[] args) {

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

        SumTask innerFind = new SumTask(src);

        long start = System.currentTimeMillis();

        /*同步提交*/
        int[] invoke = pool.invoke(innerFind);
//        for(int number:invoke){
//            System.out.println(number);
//        }
        System.out.println(" spend time:"+(System.currentTimeMillis()-start)+"ms");

    }
}

異步示例

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

    private File path;

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

    @Override
    protected void compute() {
        List<FindDirsFiles> subTasks = new ArrayList<>();

        File[] files = path.listFiles();
        if (files!=null){
            for (File file : files) {
                if (file.isDirectory()) {
                    // 對每個(gè)子目錄都新建一個(gè)子任務(wù)。
                    subTasks.add(new FindDirsFiles(file));
                } else {
                    // 遇到文件，檢查。
                    if (file.getAbsolutePath().endsWith("txt")){
                        System.out.println("文件:" + file.getAbsolutePath());
                    }
                }
            }
            if (!subTasks.isEmpty()) {
                // 在當(dāng)前的 ForkJoinPool 上調(diào)度所有的子任務(wù)。
                for (FindDirsFiles subTask : invokeAll(subTasks)) {
                    subTask.join();
                }
            }
        }
    }

    public static void main(String [] args){
        try {
            // 用一個(gè) ForkJoinPool 實(shí)例調(diào)度總?cè)蝿?wù)
            ForkJoinPool pool = new ForkJoinPool();
            FindDirsFiles task = new FindDirsFiles(new File("F:/"));

            /*異步提交*/
            pool.execute(task);

            /*主線程做自己的業(yè)務(wù)工作*/
            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) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

二、CountDownLatch

閉鎖，CountDownLatch 這個(gè)類能夠使一個(gè)線程等待其他線程完成各自的工作后再執(zhí)行。例如，應(yīng)用程序的主線程希望在負(fù)責(zé)啟動(dòng)框架服務(wù)的線程已經(jīng)啟動(dòng)所有的框架服務(wù)之后再執(zhí)行。

CountDownLatch 是通過一個(gè)計(jì)數(shù)器來實(shí)現(xiàn)的，計(jì)數(shù)器的初始值為初始任務(wù)的數(shù)量。每當(dāng)完成了一個(gè)任務(wù)后，計(jì)數(shù)器的值就會(huì)減 1（CountDownLatch.countDown()方法）。當(dāng)計(jì)數(shù)器值到達(dá) 0 時(shí)，它表示所有的已經(jīng)完成了任務(wù)，然后在閉鎖上等待 CountDownLatch.await()方法的線程就可以恢復(fù)執(zhí)行任務(wù)。

應(yīng)用場景：

實(shí)現(xiàn)最大的并行性：有時(shí)我們想同時(shí)啟動(dòng)多個(gè)線程，實(shí)現(xiàn)最大程度的并行性。例如，我們想測試一個(gè)單例類。如果我們創(chuàng)建一個(gè)初始計(jì)數(shù)為 1 的CountDownLatch，并讓所有線程都在這個(gè)鎖上等待，那么我們可以很輕松地完成測試。我們只需調(diào)用 一次 countDown()方法就可以讓所有的等待線程同時(shí)恢復(fù)執(zhí)行。

開始執(zhí)行前等待 n 個(gè)線程完成各自任務(wù)：例如應(yīng)用程序啟動(dòng)類要確保在處理用戶請求前，所有 N 個(gè)外部系統(tǒng)已經(jīng)啟動(dòng)和運(yùn)行了，例如處理 excel 中多個(gè)表單。

image

參見代碼包 cn.enjoyedu.ch2.tools 下

三、CyclicBarrier

CyclicBarrier 的字面意思是可循環(huán)使用（Cyclic）的屏障（Barrier）。它要做的事情是，讓一組線程到達(dá)一個(gè)屏障（也可以叫同步點(diǎn)）時(shí)被阻塞，直到最后一個(gè)線程到達(dá)屏障時(shí)，屏障才會(huì)開門，所有被屏障攔截的線程才會(huì)繼續(xù)運(yùn)行。CyclicBarrier 默認(rèn)的構(gòu)造方法是 CyclicBarrier（int parties），其參數(shù)表示屏障攔截的線程數(shù)量，每個(gè)線程調(diào)用 await 方法告訴 CyclicBarrier 我已經(jīng)到達(dá)了屏障，然后當(dāng)前線程被阻塞。

CyclicBarrier 還提供一個(gè)更高級的構(gòu)造函數(shù) CyclicBarrie（r int parties，RunnablebarrierAction），用于在線程到達(dá)屏障時(shí)，優(yōu)先執(zhí)行 barrierAction，方便處理更復(fù)雜的業(yè)務(wù)場景。

CyclicBarrier 可以用于多線程計(jì)算數(shù)據(jù)，最后合并計(jì)算結(jié)果的場景。

參見代碼包 cn.enjoyedu.ch2.tools 下

四、CountDownLatch 和 CyclicBarrier 辨析

CountDownLatch 的計(jì)數(shù)器只能使用一次，而 CyclicBarrier 的計(jì)數(shù)器可以反復(fù)使用。

CountDownLatch.await 一般阻塞工作線程，所有的進(jìn)行預(yù)備工作的線程執(zhí)行countDown，而 CyclicBarrier 通過工作線程調(diào)用 await 從而自行阻塞，直到所有工作線程達(dá)到指定屏障，再大家一起往下走。

在控制多個(gè)線程同時(shí)運(yùn)行上，CountDownLatch 可以不限線程數(shù)量，而CyclicBarrier 是固定線程數(shù)。同時(shí)，CyclicBarrier 還可以提供一個(gè) barrierAction，合并多線程計(jì)算結(jié)果。

五、Semaphore

Semaphore（信號(hào)量）是用來控制同時(shí)訪問特定資源的線程數(shù)量，它通過協(xié)調(diào)各個(gè)線程，以保證合理的使用公共資源。應(yīng)用場景 Semaphore 可以用于做流量控制，特別是公用資源有限的應(yīng)用場景，比如數(shù)據(jù)庫連接。假如有一個(gè)需求，要讀取幾萬個(gè)文件的數(shù)據(jù)，因?yàn)槎际?IO 密集型任務(wù)，我們可以啟動(dòng)幾十個(gè)線程并發(fā)地讀取，但是如果讀到內(nèi)存后，還需要存儲(chǔ)到數(shù)據(jù)庫中，而數(shù)據(jù)庫的連接數(shù)只有 10 個(gè)，這時(shí)我們必須控制只有 10 個(gè)線程同時(shí)獲取數(shù)據(jù)庫連接保存數(shù)據(jù)，否則會(huì)報(bào)錯(cuò)無法獲取數(shù)據(jù)庫連接。這個(gè)時(shí)候，就可以使用 Semaphore 來做流量控制。。Semaphore 的構(gòu)造方法 Semaphore（int permits）接受一個(gè)整型的數(shù)字，表示可用的許可證數(shù)量。Semaphore 的用法也很簡單，首先線程使用 Semaphore的 acquire()方法獲取一個(gè)許可證，使用完之后調(diào)用 release()方法歸還許可證。還可以用 tryAcquire()方法嘗試獲取許可證。Semaphore 還提供一些其他方法，具體如下。

?intavailablePermits()：返回此信號(hào)量中當(dāng)前可用的許可證數(shù)。

?intgetQueueLength()：返回正在等待獲取許可證的線程數(shù)。

?booleanhasQueuedThreads()：是否有線程正在等待獲取許可證。

?void reducePermit（s int reduction）：減少 reduction 個(gè)許可證，是個(gè) protected方法。

?Collection getQueuedThreads()：返回所有等待獲取許可證的線程集合，是個(gè) protected 方法。

1、用 Semaphore 實(shí)現(xiàn)數(shù)據(jù)庫連接池

參見代碼,包 cn.enjoyedu.ch2.tools.semaphore 下

2、Semaphore 注意事項(xiàng)

參見代碼類 cn.enjoyedu.ch2.tools.semaphore. DBPoolNoUseless 下

六、Exchange

Exchanger（交換者）是一個(gè)用于線程間協(xié)作的工具類。Exchanger 用于進(jìn)行線程間的數(shù)據(jù)交換。它提供一個(gè)同步點(diǎn)，在這個(gè)同步點(diǎn)，兩個(gè)線程可以交換彼此的數(shù)據(jù)。這兩個(gè)線程通過 exchange 方法交換數(shù)據(jù)，如果第一個(gè)線程先執(zhí)行exchange()方法，它會(huì)一直等待第二個(gè)線程也執(zhí)行 exchange 方法，當(dāng)兩個(gè)線程都到達(dá)同步點(diǎn)時(shí)，這兩個(gè)線程就可以交換數(shù)據(jù)，將本線程生產(chǎn)出來的數(shù)據(jù)傳遞給對方。

參見代碼包 cn.enjoyedu.ch2.tools 下

七、Callable、Future 和 FutureTask

Runnable 是一個(gè)接口，在它里面只聲明了一個(gè) run()方法，由于 run()方法返回值為 void 類型，所以在執(zhí)行完任務(wù)之后無法返回任何結(jié)果。

Callable 位于 java.util.concurrent 包下，它也是一個(gè)接口，在它里面也只聲明了一個(gè)方法，只不過這個(gè)方法叫做 call()，這是一個(gè)泛型接口，call()函數(shù)返回的類型就是傳遞進(jìn)來的 V 類型。

Future 就是對于具體的 Runnable 或者 Callable 任務(wù)的執(zhí)行結(jié)果進(jìn)行取消、查詢是否完成、獲取結(jié)果。必要時(shí)可以通過 get 方法獲取執(zhí)行結(jié)果，該方法會(huì)阻塞直到任務(wù)返回結(jié)果。

image

因?yàn)?Future 只是一個(gè)接口，所以是無法直接用來創(chuàng)建對象使用的，因此就有了下面的 FutureTask。

image

FutureTask 類實(shí)現(xiàn)了 RunnableFuture 接口，RunnableFuture 繼承了 Runnable接口和 Future 接口，而 FutureTask 實(shí)現(xiàn)了 RunnableFuture 接口。所以它既可以作為 Runnable 被線程執(zhí)行，又可以作為 Future 得到 Callable 的返回值。

image

因此我們通過一個(gè)線程運(yùn)行 Callable，但是 Thread 不支持構(gòu)造方法中傳遞Callable 的實(shí)例，所以我們需要通過 FutureTask 把一個(gè) Callable 包裝成 Runnable，然后再通過這個(gè) FutureTask 拿到Callable 運(yùn)行后的返回值。

要 new 一個(gè) FutureTask 的實(shí)例，有兩種方法

image

參見代碼包 cn.enjoyedu.ch2.tools 下