ThreadPool

本文原創(chuàng)地址，我的博客：https://jsbintask.cn/2019/03/10/jdk/jdk8-threadpool/(食用效果最佳)，轉(zhuǎn)載請(qǐng)注明出處!

前言

在實(shí)際工作中，線(xiàn)程是一個(gè)我們經(jīng)常要打交道的角色，它可以幫我們靈活利用資源，提升程序運(yùn)行效率。但是我們今天不是探討線(xiàn)程！我們今天來(lái)聊聊另一個(gè)與線(xiàn)程息息相關(guān)的角色：線(xiàn)程池.本篇文章的目的就是全方位的解析線(xiàn)程池的作用，以及jdk中的接口，實(shí)現(xiàn)以及原理，另外對(duì)于某些重要概念，將從源碼的角度探討。
tip：本文較長(zhǎng)，建議先碼后看。

線(xiàn)程池介紹

首先我們看一段創(chuàng)建線(xiàn)程并且運(yùn)行的常用代碼：

for (int i = 0; i < 100; i++) {
    new Thread(() -> {
        System.out.println("run thread->" + Thread.currentThread().getName());
        //to do something, send email, message, io operator, network...
    }).start();
}

上面的代碼很容易理解，我們?yōu)榱水惒剑蛘咝士紤]，將某些耗時(shí)操作放入一個(gè)新線(xiàn)程去運(yùn)行，但是這樣的代碼卻存在這樣的問(wèn)題：

1. 創(chuàng)建銷(xiāo)毀線(xiàn)程資源消耗； 我們使用線(xiàn)程的目的本是出于效率考慮，可以為了創(chuàng)建這些線(xiàn)程卻消耗了額外的時(shí)間，資源，對(duì)于線(xiàn)程的銷(xiāo)毀同樣需要系統(tǒng)資源。
2. cpu資源有限，上述代碼創(chuàng)建線(xiàn)程過(guò)多，造成有的任務(wù)不能即時(shí)完成，響應(yīng)時(shí)間過(guò)長(zhǎng)。
3. 線(xiàn)程無(wú)法管理，無(wú)節(jié)制地創(chuàng)建線(xiàn)程對(duì)于有限的資源來(lái)說(shuō)似乎成了“得不償失”的一種作用。
   手動(dòng)創(chuàng)建執(zhí)行線(xiàn)程存在以上問(wèn)題，而線(xiàn)程池就是用來(lái)解決這些問(wèn)題的。怎么解決呢？我們可以先粗略的定義一下線(xiàn)程池：

線(xiàn)程池是一組已經(jīng)創(chuàng)建好的，一直在等待任務(wù)執(zhí)行的線(xiàn)程的集合。

因?yàn)榫€(xiàn)程池中線(xiàn)程是已經(jīng)創(chuàng)建好的，所以對(duì)于任務(wù)的執(zhí)行不會(huì)消耗掉額外的資源，線(xiàn)程池中線(xiàn)程個(gè)數(shù)由我們自定義添加，可相對(duì)于資源，資源任務(wù)做出調(diào)整，對(duì)于某些任務(wù)，如果線(xiàn)程池尚未執(zhí)行，可手動(dòng)取消，線(xiàn)程任務(wù)變得能夠管理！
所以，線(xiàn)程池的作用如下：

1. 降低資源消耗。通過(guò)重復(fù)利用已創(chuàng)建的線(xiàn)程降低線(xiàn)程創(chuàng)建和銷(xiāo)毀造成的消耗。
2. 提高響應(yīng)速度。當(dāng)任務(wù)到達(dá)時(shí)，任務(wù)可以不需要等到線(xiàn)程創(chuàng)建就能立即執(zhí)行。
3. 提高線(xiàn)程的可管理性。

jdk線(xiàn)程池詳解

上面我們已經(jīng)知道了線(xiàn)程池的作用，而對(duì)于這樣一個(gè)好用，重要的工具，jdk當(dāng)然已經(jīng)為我們提供了實(shí)現(xiàn)，這也是本篇文章的重點(diǎn)。
在jdk中，關(guān)于線(xiàn)程池的接口，類(lèi)都定義在juc（java.util.concurrent）包中，這是jdk專(zhuān)門(mén)為我們提供用于并發(fā)編程的包，當(dāng)然，本篇文章我們只介紹與線(xiàn)程池有關(guān)的接口和類(lèi)，首先我們看下重點(diǎn)要學(xué)習(xí)的接口和類(lèi)：

ThreadPool

Executor

首先我們需要了解就是Executor接口，它有一個(gè)方法，定義如下：

ThreadPool

ExecutorService

[圖片上傳失敗...(image-3bf934-1552379783669)]
這個(gè)接口繼承自Executor，它的方法定義就豐富多了，可以關(guān)閉，提交Future任務(wù)，批量提交任務(wù)，獲取執(zhí)行結(jié)果等，我們一一講解下每個(gè)方法作用聲明：

1. void shutdown(): “優(yōu)雅地”關(guān)閉線(xiàn)程池，為什么是“優(yōu)雅地”呢？因?yàn)檫@個(gè)線(xiàn)程池在關(guān)閉前會(huì)先等待線(xiàn)程池中已經(jīng)有的任務(wù)執(zhí)行完成，一般會(huì)配合方法awaitTermination一起使用，調(diào)用該方法后，線(xiàn)程池中不能再加入新的任務(wù)。
2. List<Runnable> shutdownNow();: “嘗試”終止正在執(zhí)行的線(xiàn)程，返回在正在等待的任務(wù)列表，調(diào)用這個(gè)方法后，會(huì)調(diào)用正在執(zhí)行線(xiàn)程的interrupt（）方法，所以如果正在執(zhí)行的線(xiàn)程如果調(diào)用了sleep，join，await等方法，會(huì)拋出InterruptedException異常。
3. boolean awaitTermination(long timeout, TimeUnit unit): 該方法是一個(gè)阻塞方法，參數(shù)分別為時(shí)間和時(shí)間單位。這個(gè)方法一般配合上面兩個(gè)方法之后調(diào)用。如果先調(diào)用shutdown方法，所有任務(wù)執(zhí)行完成返回true，超時(shí)返回false，如果先調(diào)用的是shutdownNow方法，正在執(zhí)行的任務(wù)全部完成true，超時(shí)返回false。
4. boolean isTerminated();： 調(diào)用方法1或者2后，如果所有人物全部執(zhí)行完畢則返回true，也就是說(shuō)，就算所有任務(wù)執(zhí)行完畢，但是不是先調(diào)用1或者2，也會(huì)返回false。
5. <T> Future<T> submit(Callable<T> task);: 提交一個(gè)能夠返回結(jié)果的Callable任務(wù)，返回任務(wù)結(jié)果抽象對(duì)象是Future，調(diào)用Future.get()方法可以阻塞等待獲取執(zhí)行結(jié)果，例如：
   result = exec.submit(aCallable).get();，提交一個(gè)任務(wù)并且一直阻塞知道該任務(wù)執(zhí)行完成獲取到返回結(jié)果。
6. <T> Future<T> submit(Runnable task, T result);： 提交一個(gè)Runnable任務(wù)，執(zhí)行成功后調(diào)用Future.get()方法返回的是result（這是什么騷操作？）。
7. Future<?> submit(Runnable task);：和6不同的是調(diào)用Future.get（）方法返回的是null（這又是什么操作？）。
8. <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks): 提交一組任務(wù)，并且返回每個(gè)任務(wù)執(zhí)行結(jié)果的抽象對(duì)象List<Future<T>>，F(xiàn)uture作用同上，值得注意的是：
   當(dāng)調(diào)用其中任一Future.isDone()(判斷任務(wù)是否完成，正常，異常終止都算）方法時(shí)，必須等到所有任務(wù)都完成時(shí)才返回true，簡(jiǎn)單說(shuō)：全部任務(wù)完成才算完成。
9. <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit): 同方法8，多了一個(gè)時(shí)間參數(shù)，不同的是：如果超時(shí)，F(xiàn)uture.isDone()同樣返回true。
10. <T> T invokeAny(Collection<? extends Callable<T>> tasks)：這個(gè)看名字和上面對(duì)比就容易理解了，返回第一個(gè)正常完成的任務(wù)地執(zhí)行結(jié)果，后面沒(méi)有完成的任務(wù)將被取消。
11. <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)：同10相比，多了一個(gè)超時(shí)參數(shù)。不同的是：在超時(shí)時(shí)間內(nèi)，一個(gè)任務(wù)都沒(méi)有完成，將拋出TimeoutException。
    到現(xiàn)在，我們已經(jīng)知道了一個(gè)線(xiàn)程池基本的所有方法，知道了每個(gè)方法的作用，接下來(lái)我們就來(lái)看看具體實(shí)現(xiàn)，首先我們研究下ExecutorService的具體實(shí)現(xiàn)抽象類(lèi)：AbstractExecutorService。

AbstractExecutorService

ThreadPool

1. submit方法：

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}

• 判空
• 利用task構(gòu)建一個(gè)Future的子類(lèi)RunnableFuture，最后返回
• 執(zhí)行這個(gè)任務(wù)（execute方法聲明在Executor接口中，所以也是交由子類(lèi)實(shí)現(xiàn))。
  execute方法交由子類(lèi)實(shí)現(xiàn)了，這里我們主要分析newTaskFor方法，看它是如何構(gòu)建Future對(duì)象的：
  首先，RunnableFuture接口定義如下：

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

他就是Future和Runnable的組合，它的實(shí)現(xiàn)是FutureTask：

ThreadPool

2. invokeAll方法：

public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
        if (tasks == null)
            throw new NullPointerException();
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;  // ①
        try {
            for (Callable<T> t : tasks) {  // ②
                RunnableFuture<T> f = newTaskFor(t);
                futures.add(f);
                execute(f);
            }
            for (int i = 0, size = futures.size(); i < size; i++) {
                Future<T> f = futures.get(i);    // ③
                if (!f.isDone()) {
                    try {
                        f.get();
                    } catch (CancellationException ignore) {
                    } catch (ExecutionException ignore) {
                    }
                }
            }
            done = true;   //  ④
            return futures;
        } finally {
            if (!done)     //   ⑤
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
        }
}

1. 聲明一個(gè)flag判斷所有任務(wù)是否全部完成
2. 調(diào)用newTaskFor方法構(gòu)建RunnableFuture對(duì)象，循環(huán)調(diào)用execute方法添加每一個(gè)任務(wù)。
3. 遍歷每個(gè)任務(wù)結(jié)果，判斷是否執(zhí)行完成，沒(méi)有完成調(diào)用 get()阻塞方法等待完成。
4. 所有任務(wù)全部完成，將flag設(shè)置成true。
5. 出現(xiàn)異常，還有任務(wù)沒(méi)有完成，所有任務(wù)取消：Future.cancel()（實(shí)際是調(diào)用執(zhí)行線(xiàn)程的interrupt方法。
   上面代碼分析和我們一開(kāi)始講解ExecutorService的invokeAll一致。

3. invokeAny方法

ThreadPool

private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
                              boolean timed, long nanos)
        throws InterruptedException, ExecutionException, TimeoutException {
        if (tasks == null)
            throw new NullPointerException();
        int ntasks = tasks.size();
        if (ntasks == 0)
            throw new IllegalArgumentException();
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
        ExecutorCompletionService<T> ecs =     // ①
            new ExecutorCompletionService<T>(this);

        try {
            ExecutionException ee = null;
            final long deadline = timed ? System.nanoTime() + nanos : 0L;
            Iterator<? extends Callable<T>> it = tasks.iterator();
            
            futures.add(ecs.submit(it.next()));       // ②
            --ntasks;
            int active = 1;

            for (;;) {
                Future<T> f = ecs.poll();    //  ③
                if (f == null) {
                    if (ntasks > 0) {
                        --ntasks;
                        futures.add(ecs.submit(it.next()));
                        ++active;
                    }
                    else if (active == 0)
                        break;
                    else if (timed) {
                        f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                        if (f == null)
                            throw new TimeoutException();
                        nanos = deadline - System.nanoTime();
                    }
                    else                  //  ④
                        f = ecs.take();
                }
                if (f != null) {           // ⑤
                    --active;
                    try {
                        return f.get();
                    } catch (ExecutionException eex) {
                        ee = eex;
                    } catch (RuntimeException rex) {
                        ee = new ExecutionException(rex);
                    }
                }
            }

            if (ee == null)       
                ee = new ExecutionException();
            throw ee;

        } finally {
            for (int i = 0, size = futures.size(); i < size; i++)      //  ⑥
                futures.get(i).cancel(true);
        }
    }

1. 聲明一個(gè)ExecutorCompletionService ecs，這個(gè)對(duì)象實(shí)際是一個(gè)任務(wù)執(zhí)行結(jié)果阻塞隊(duì)列和線(xiàn)程池的結(jié)合，所以它可以加入任務(wù)，執(zhí)行任務(wù)，將任務(wù)執(zhí)行結(jié)果加入阻塞隊(duì)列。
2. 向ecs添加tasks中的第一個(gè)任務(wù)并且執(zhí)行。
3. 從ecs的阻塞隊(duì)列中取出第一個(gè)（隊(duì)頭），如果為null（不為null跳到注釋⑤），說(shuō)明一個(gè)任務(wù)都還沒(méi)執(zhí)行完成，繼續(xù)添加任務(wù)。
4. 如果所有任務(wù)都被添加了，阻塞等待任務(wù)的執(zhí)行結(jié)果，知道有任一任務(wù)執(zhí)行完成。
5. 如果取到了某個(gè)任務(wù)的執(zhí)行結(jié)果，直接返回。
6. 取消所有還沒(méi)執(zhí)行的任務(wù)。
   上面代碼分析和我們一開(kāi)始講解ExecutorService的invokeAny一致。 到現(xiàn)在，我們已經(jīng)分析完了AbstractExecutorService中的公共的方法，接下來(lái)就該研究最終的具體實(shí)現(xiàn)了：ThreadPoolExecutor

ThreadPoolExecutor

ThreadPoolExecutor繼承自AbstractExecutorService,它是線(xiàn)程池的具體實(shí)現(xiàn)：

ThreadPool

ThreadPool

1. corePoolSize=maximuPoolSize，我們可以創(chuàng)建一個(gè)線(xiàn)程池線(xiàn)程數(shù)量固定的任務(wù)。
2. maximumPoolSize設(shè)置的足夠大（Integer.MAX_VALUE），可以無(wú)限制的加入任務(wù)。
3. workQueue設(shè)置的足夠大，線(xiàn)程池中的數(shù)量不會(huì)超過(guò)corePoolSize，此時(shí)maximumPoolSize參數(shù)無(wú)用。
4. corePoolSize=0，線(xiàn)程池一旦空閑（超過(guò)時(shí)間），線(xiàn)程都將被回收。
5. 我們上面知道，如果多余的空閑線(xiàn)程空閑時(shí)間超過(guò)keepAliveTime*unit，這些線(xiàn)程將被回收。我們可以通過(guò)方法allowCoreThreadTimeOut使這個(gè)參數(shù)對(duì)線(xiàn)程池中所有線(xiàn)程都有效果。
6. workQueue一般有三種實(shí)現(xiàn)：

• SynchronousQueue，這是一個(gè)空隊(duì)列，不會(huì)保存提交的task（添加操作必須等待另外的移除操作）。
• ArrayBlockingQueue，數(shù)組實(shí)現(xiàn)的丟列，可以指定隊(duì)列的長(zhǎng)度。
• LinkedBlockingQueue, 鏈表實(shí)現(xiàn)的隊(duì)列，所以理論上可以無(wú)限大，也可以指定鏈表長(zhǎng)度。

7. 而RejectedExecutionHandler一般由四種實(shí)現(xiàn)：

• AbortPolicy, 直接拋出RejectedExecutionException，這是線(xiàn)程池中的默認(rèn)實(shí)現(xiàn)
• DiscardPolicy，什么都不做
• DiscardOldestPolicy，丟棄workQueue隊(duì)頭任務(wù)，加入新任務(wù)
• CallerRunsPolicy，直接在調(diào)用者的線(xiàn)程執(zhí)行任務(wù)
  最后，我們?cè)俜治鱿耇hreadPoolExecutor核心方法execute：

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        
        int c = ctl.get();   // ①
        if (workerCountOf(c) < corePoolSize) {   // ②
            if (addWorker(command, true))
                return;
            c = ctl.get();     // ③
        }
        if (isRunning(c) && workQueue.offer(command)) {    // ④
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))   // ⑤
                reject(command);
            else if (workerCountOf(recheck) == 0)    // ⑥
                addWorker(null, false);
        }
        else if (!addWorker(command, false))    // ⑦
            reject(command);
    }

1. 獲取線(xiàn)程池中的線(xiàn)程數(shù)量
2. 線(xiàn)程池中線(xiàn)程數(shù)量小于corePoolSize，直接調(diào)用addWorker添加新線(xiàn)程執(zhí)行任務(wù)返回。
3. 因?yàn)槎嗑€(xiàn)程的關(guān)系，上一步可能調(diào)用addWorker失敗（其它線(xiàn)程創(chuàng)建了，數(shù)以數(shù)量已經(jīng)超過(guò)了），重啟獲取線(xiàn)程數(shù)量。
4. 向workQueue添加添加任務(wù)，如果添加成功，double獲取線(xiàn)程數(shù)量，添加失敗，走到步驟⑦
5. double檢查后發(fā)現(xiàn)線(xiàn)程池已經(jīng)關(guān)閉或者數(shù)量超出，回滾已經(jīng)添加的任務(wù)（remove（command））并且執(zhí)行拒絕策略。
6. double檢查通過(guò)，添加一個(gè)新線(xiàn)程。

7. 再次添加線(xiàn)程，失敗則調(diào)用拒絕策略。
   好了，到現(xiàn)在jdk中的線(xiàn)程池核心的實(shí)現(xiàn)，策略，分析我們已經(jīng)分析完成了。接下來(lái)我我們就來(lái)看看關(guān)于線(xiàn)程池的另外的一些擴(kuò)展，也就是圖上的剩下的接口和類(lèi)：

   
   
   ThreadPool

ScheduledExecutorService

ScheduledExecutorService繼承自ExecutorService，ExecutorService的分析上面我們已經(jīng)知道了，我們來(lái)看看它擴(kuò)展了哪些方法：

ThreadPool

1. public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);: 這個(gè)方法用于定時(shí)執(zhí)行任務(wù)command，延遲的時(shí)間為delay*unit，它返回一個(gè)ScheduledFuture對(duì)象用于獲取執(zhí)行結(jié)果或者剩余延時(shí)，調(diào)用Future.get()方法將阻塞當(dāng)前線(xiàn)程最后返回null。
   
   ThreadPool
2. public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);：同上，不同的是，調(diào)用Future.get()方法將返回執(zhí)行的結(jié)果，而不是null。
3. public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period,TimeUnit unit);: 重復(fù)執(zhí)行任務(wù)command，第一次執(zhí)行時(shí)間為initialDelay延遲后，以后的執(zhí)行時(shí)間將在initialDelay + period * n，unit代表時(shí)間單位，值得注意的是，如果某次執(zhí)行出現(xiàn)異常，后面該任務(wù)就不會(huì)再執(zhí)行?；蛘咄ㄟ^(guò)返回對(duì)象Future手動(dòng)取消，后面也將不再執(zhí)行。
4. public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,long initialDelay,long delay, TimeUnit unit);: 效果同上，不同點(diǎn)：如果command耗時(shí)為 y，則上面的計(jì)算公式為initialDelay + period * n + y，也就是說(shuō)，它的定時(shí)時(shí)間會(huì)加上任務(wù)耗時(shí)，而上面的方法則是一個(gè)固定的頻率，不會(huì)算上任務(wù)執(zhí)行時(shí)間！
   這是它擴(kuò)展的四個(gè)方法，其中需要注意的是scheduleAtFixedRate和scheduleWithFixedDelay的細(xì)微差別，最后，我們來(lái)看下它的實(shí)現(xiàn)類(lèi)：ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor繼承自ThreadPoolExecutor類(lèi)，實(shí)現(xiàn)了ScheduledExecutorService接口，上面均已經(jīng)分析。

ThreadPool

ThreadPool

ThreadPool

 public RunnableScheduledFuture<?> take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                for (;;) {    // ①
                    RunnableScheduledFuture<?> first = queue[0];
                    if (first == null)
                        available.await();
                    else {
                        long delay = first.getDelay(NANOSECONDS);
                        if (delay <= 0)
                            return finishPoll(first);
                        first = null; // don't retain ref while waiting
                        if (leader != null)
                            available.await();
                        else {
                            Thread thisThread = Thread.currentThread();
                            leader = thisThread;
                            try {
                                available.awaitNanos(delay);
                            } finally {
                                if (leader == thisThread)
                                    leader = null;
                            }
                        }
                    }
                }
            } finally {
                if (leader == null && queue[0] != null)
                    available.signal();
                lock.unlock();
            }
        }

1. 工作線(xiàn)程調(diào)用take方法獲取剩余任務(wù)。
2. 檢查這個(gè)任務(wù)是否已經(jīng)到了執(zhí)行時(shí)間。
3. 未到執(zhí)行時(shí)間，await等待。
4. 自己?jiǎn)拘眩M(jìn)入循環(huán)再次計(jì)算時(shí)間。
   好了，到目前為止jdk中關(guān)于線(xiàn)程池的6個(gè)核心類(lèi)已經(jīng)全部分析完畢了。接下來(lái)還有最后一個(gè)小問(wèn)題，我們手動(dòng)創(chuàng)建線(xiàn)程池參數(shù)也太了，不管是ThreadPoolExecutor還是ScheduledThreadPoolExecutor，這對(duì)于用戶(hù)來(lái)說(shuō)似乎并不太友好，當(dāng)然，jdk已經(jīng)想到了這個(gè)問(wèn)題，所以，我們最后再介紹一個(gè)創(chuàng)建這些線(xiàn)程池的工具類(lèi)：Executors:

Executors

它的主要工具方法如下：

ThreadPool

ThreadPool

總結(jié)

本次，我們首先從代碼出發(fā)，分析了線(xiàn)程池給我們帶來(lái)的好處以及直接使用線(xiàn)程的弊端，接著引出了jdk中的已經(jīng)實(shí)現(xiàn)了的線(xiàn)程池。然后重點(diǎn)分析了jdk中關(guān)于線(xiàn)程池的六個(gè)最重要的接口和類(lèi)，并且從源碼角度講解了關(guān)鍵點(diǎn)實(shí)現(xiàn)，最后，處于方便考慮，我們還知道jdk給我們留了一個(gè)創(chuàng)建線(xiàn)程池的工具類(lèi)，簡(jiǎn)化了手動(dòng)創(chuàng)建線(xiàn)程池的步驟。
真正做到了知其然，知其所以然。

關(guān)注我，這里只有干貨！