徹底搞懂java并發(fā)ThreadPoolExecutor使用

更新時間：2023年02月28日 15:32:01 作者：半夏之沫

這篇文章主要為大家介紹了徹底搞懂java并發(fā)ThreadPoolExecutor使用詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

線程池是Java中使用較多的并發(fā)框架，合理使用線程池，可以：降低資源消耗，提高響應(yīng)速度，提高線程的可管理性。

本篇文章將從線程池簡單原理，線程池的創(chuàng)建，線程池執(zhí)行任務(wù)和關(guān)閉線程池進(jìn)行使用學(xué)習(xí)。

正文

一. 線程池的簡單原理

當(dāng)一個任務(wù)提交到線程池ThreadPoolExecutor時，該任務(wù)的執(zhí)行如下圖所示。

如果當(dāng)前運(yùn)行的線程數(shù)小于corePoolSzie（核心線程數(shù)），則創(chuàng)建新線程來執(zhí)行任務(wù)（需要獲取全局鎖）；
如果當(dāng)前運(yùn)行的線程數(shù)等于或大于corePoolSzie，則將任務(wù)加入BlockingQueue（任務(wù)阻塞隊(duì)列）；
如果BlockingQueue已滿，則創(chuàng)建新的線程來執(zhí)行任務(wù)（需要獲取全局鎖）；
如果創(chuàng)建新線程會使當(dāng)前線程數(shù)大于maximumPoolSize（最大線程數(shù)），則拒絕任務(wù)并調(diào)用RejectedExecutionHandler的rejectedExecution() 方法。

由于ThreadPoolExecutor存儲工作線程使用的集合是HashSet，因此執(zhí)行上述步驟1和步驟3時需要獲取全局鎖來保證線程安全，而獲取全局鎖會導(dǎo)致線程池性能瓶頸，因此通常情況下，線程池完成預(yù)熱后（當(dāng)前線程數(shù)大于等于corePoolSize），線程池的execute() 方法都是執(zhí)行步驟2。

二. 線程池的創(chuàng)建

通過ThreadPoolExecutor能夠創(chuàng)建一個線程池，ThreadPoolExecutor的構(gòu)造函數(shù)簽名如下。

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue)
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory)
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          RejectedExecutionHandler handler)
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

通過ThreadPoolExecutor創(chuàng)建線程池時，需要指定線程池的核心線程數(shù)，最大線程數(shù)，線程?；顣r間，線程保活時間單位和任務(wù)阻塞隊(duì)列，并按需指定線程工廠和飽和拒絕策略，如果不指定線程工廠和飽和拒絕策略，則ThreadPoolExecutor會使用默認(rèn)的線程工廠和飽和拒絕策略。下面分別介紹這些參數(shù)的含義。

參數(shù)	含義
corePoolSize	核心線程數(shù)，即線程池的基本大小。當(dāng)一個任務(wù)被提交到線程池時，如果線程池的線程數(shù)小于corePoolSize，那么無論其余線程是否空閑，也需創(chuàng)建一個新線程來執(zhí)行任務(wù)。
maximumPoolSize	最大線程數(shù)。當(dāng)線程池中線程數(shù)大于等于corePoolSize時，新提交的任務(wù)會加入任務(wù)阻塞隊(duì)列，但是如果任務(wù)阻塞隊(duì)列已滿且線程數(shù)小于maximumPoolSize，此時會繼續(xù)創(chuàng)建新的線程來執(zhí)行任務(wù)。該參數(shù)規(guī)定了線程池允許創(chuàng)建的最大線程數(shù)
keepAliveTime	線程?；顣r間。當(dāng)線程池的線程數(shù)大于核心線程數(shù)時，多余的空閑線程會最大存活keepAliveTime的時間，如果超過這個時間且空閑線程還沒有獲取到任務(wù)來執(zhí)行，則該空閑線程會被回收掉。
unit	線程保活時間單位。通過TimeUnit指定線程保活時間的時間單位，可選單位有DAYS（天），HOURS（時），MINUTES（分），SECONDS（秒），MILLISECONDS（毫秒），MICROSECONDS（微秒）和NANOSECONDS（納秒），但無論指定什么時間單位，ThreadPoolExecutor統(tǒng)一會將其轉(zhuǎn)換為NANOSECONDS。
workQueue	任務(wù)阻塞隊(duì)列。線程池的線程數(shù)大于等于corePoolSize時，新提交的任務(wù)會添加到workQueue中，所有線程執(zhí)行完上一個任務(wù)后，會循環(huán)從workQueue中獲取任務(wù)來執(zhí)行。
threadFactory	創(chuàng)建線程的工廠?？梢酝ㄟ^線程工廠給每個創(chuàng)建出來的線程設(shè)置更有意義的名字。
handler	飽和拒絕策略。如果任務(wù)阻塞隊(duì)列已滿且線程池中的線程數(shù)等于maximumPoolSize，說明線程池此時處于飽和狀態(tài)，應(yīng)該執(zhí)行一種拒絕策略來處理新提交的任務(wù)。

三. 線程池執(zhí)行任務(wù)

1. 執(zhí)行無返回值任務(wù)

通過ThreadPoolExecutor的execute() 方法，能執(zhí)行Runnable任務(wù)，示例如下。

public class ThreadPoolExecutorTest {
    @Test
    public void ThreadPoolExecutor執(zhí)行簡單無返回值任務(wù)() throws Exception {
        // 創(chuàng)建一個線程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                60, TimeUnit.SECONDS, new ArrayBlockingQueue&lt;&gt;(300));
        // 創(chuàng)建兩個任務(wù)
        Runnable firstRunnable = new Runnable() {
            @Override
            public void run() {
                System.out.println("第一個任務(wù)執(zhí)行");
            }
        };
        Runnable secondRunnable = new Runnable() {
            @Override
            public void run() {
                System.out.println("第二個任務(wù)執(zhí)行");
            }
        };
        // 讓線程池執(zhí)行任務(wù)
        threadPoolExecutor.execute(firstRunnable);
        threadPoolExecutor.execute(secondRunnable);
        // 讓主線程睡眠1秒，等待線程池中的任務(wù)被執(zhí)行完畢
        Thread.sleep(1000);
    }
}

運(yùn)行測試程序，結(jié)果如下。

2. 執(zhí)行有返回值任務(wù)

通過ThreadPoolExecutor的submit() 方法，能夠執(zhí)行Callable任務(wù)，通過submit() 方法返回的RunnableFuture能夠拿到異步執(zhí)行的結(jié)果。示例如下。

public class ThreadPoolExecutorTest {
    @Test
    public void ThreadPoolExecutor執(zhí)行簡單有返回值任務(wù)() throws Exception {
        // 創(chuàng)建一個線程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                60, TimeUnit.SECONDS, new ArrayBlockingQueue&lt;&gt;(300));
        // 創(chuàng)建兩個任務(wù)，任務(wù)執(zhí)行完有返回值
        Callable&lt;String&gt; firstCallable = new Callable&lt;String&gt;() {
            @Override
            public String call() throws Exception {
                return "第一個任務(wù)返回值";
            }
        };
        Callable&lt;String&gt; secondCallable = new Callable&lt;String&gt;() {
            @Override
            public String call() throws Exception {
                return "第二個任務(wù)返回值";
            }
        };
        // 讓線程池執(zhí)行任務(wù)
        Future&lt;String&gt; firstFuture = threadPoolExecutor.submit(firstCallable);
        Future&lt;String&gt; secondFuture = threadPoolExecutor.submit(secondCallable);
        // 獲取執(zhí)行結(jié)果，拿不到結(jié)果會阻塞在get()方法上
        System.out.println(firstFuture.get());
        System.out.println(secondFuture.get());
    }
}

運(yùn)行測試程序，結(jié)果如下。

3. 執(zhí)行有返回值任務(wù)時拋出錯誤

如果ThreadPoolExecutor在執(zhí)行Callable任務(wù)時，在Callable任務(wù)中拋出了異常并且沒有捕獲，那么這個異常是可以通過Future的get() 方法感知到的。示例如下。

public class ThreadPoolExecutorTest {
    @Test
    public void ThreadPoolExecutor執(zhí)行簡單有返回值任務(wù)時拋出錯誤() {
        // 創(chuàng)建一個線程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
        // 創(chuàng)建一個任務(wù)，任務(wù)有返回值，但是執(zhí)行過程中拋出異常
        Callable<String> exceptionCallable = new Callable<String>() {
            @Override
            public String call() throws Exception {
                throw new RuntimeException("發(fā)生了異常");
            }
        };
        // 讓線程池執(zhí)行任務(wù)
        Future<String> exceptionFuture = threadPoolExecutor.submit(exceptionCallable);
        try {
            System.out.println(exceptionFuture.get());
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
    }
}

運(yùn)行測試程序，結(jié)果如下。

4. ThreadPoolExecutor通過submit方式執(zhí)行Runnable

ThreadPoolExecutor可以通過submit() 方法來運(yùn)行Runnable任務(wù)，并且還可以異步獲取執(zhí)行結(jié)果。示例如下。

public class ThreadPoolExecutorTest {
    @Test
    public void ThreadPoolExecutor通過submit的方式來提交并執(zhí)行Runnable() throws Exception {
        // 創(chuàng)建一個線程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                60, TimeUnit.SECONDS, new ArrayBlockingQueue&lt;&gt;(300));
        // 創(chuàng)建結(jié)果對象
        MyResult myResult = new MyResult();
        // 創(chuàng)建Runnable對象
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                myResult.setResult("任務(wù)執(zhí)行了");
            }
        };
        // 通過ThreadPoolExecutor的submit()方法提交Runnable
        Future&lt;MyResult&gt; resultFuture = threadPoolExecutor.submit(runnable, myResult);
        // 獲取執(zhí)行結(jié)果
        MyResult finalResult = resultFuture.get();
        // myResult和finalResult的地址實(shí)際相同
        Assert.assertEquals(myResult, finalResult);
        // 打印執(zhí)行結(jié)果
        System.out.println(resultFuture.get().getResult());
    }
    private static class MyResult {
        String result;
        public MyResult() {}
        public MyResult(String result) {
            this.result = result;
        }
        public String getResult() {
            return result;
        }
        public void setResult(String result) {
            this.result = result;
        }
    }
}

運(yùn)行測試程序，結(jié)果如下。

實(shí)際上ThreadPoolExecutor的submit() 方法無論是提交Runnable任務(wù)還是Callable任務(wù)，都是將任務(wù)封裝成了RunnableFuture接口的子類FutureTask，然后調(diào)用ThreadPoolExecutor的execute() 方法來執(zhí)行FutureTask。

四. 關(guān)閉線程池

關(guān)閉線程池可以通過ThreadPoolExecutor的shutdown() 方法，但是shutdown() 方法不會去中斷正在執(zhí)行任務(wù)的線程，所以如果線程池里有Worker正在執(zhí)行一個永遠(yuǎn)不會結(jié)束的任務(wù)，那么shutdown() 方法是無法關(guān)閉線程池的。示例如下。

public class ThreadPoolExecutorTest {
    @Test
    public void 通過shutdown關(guān)閉線程池() {
        // 創(chuàng)建一個線程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
        // 創(chuàng)建Runnable對象
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                while (!Thread.currentThread().isInterrupted()) {
                    LockSupport.parkNanos(1000 * 1000 * 1000);
                }
                System.out.println(Thread.currentThread().getName() + " 被中斷");
            }
        };
        // 讓線程池執(zhí)行任務(wù)
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        // 調(diào)用shutdown方法關(guān)閉線程池
        threadPoolExecutor.shutdown();
        // 等待3秒觀察現(xiàn)象
        LockSupport.parkNanos(1000 * 1000 * 1000 * 3L);
    }
}

運(yùn)行測試程序，會發(fā)現(xiàn)在主線程中等待3秒后，也沒有得到預(yù)期的打印結(jié)果。如果上述測試程序中使用shutdownNow，則是可以得到預(yù)期打印結(jié)果的，示例如下。

public class ThreadPoolExecutorTest {
    @Test
    public void 通過shutdownNow關(guān)閉線程池() {
        // 創(chuàng)建一個線程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
        // 創(chuàng)建Runnable對象
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                while (!Thread.currentThread().isInterrupted()) {
                    LockSupport.parkNanos(1000 * 1000 * 1000);
                }
                System.out.println(Thread.currentThread().getName() + " 被中斷");
            }
        };
        // 讓線程池執(zhí)行任務(wù)
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        // 調(diào)用shutdown方法關(guān)閉線程池
        threadPoolExecutor.shutdownNow();
        // 等待3秒觀察現(xiàn)象
        LockSupport.parkNanos(1000 * 1000 * 1000 * 3L);
    }
}

運(yùn)行測試程序，打印如下。

因?yàn)闇y試程序中的任務(wù)是響應(yīng)中斷的，而ThreadPoolExecutor的shutdownNow() 方法會中斷所有Worker，所以執(zhí)行shutdownNow() 方法后，正在運(yùn)行的任務(wù)會響應(yīng)中斷并結(jié)束運(yùn)行，最終線程池關(guān)閉。

假如線程池中運(yùn)行著一個永遠(yuǎn)不會結(jié)束的任務(wù)，且這個任務(wù)不響應(yīng)中斷，那么無論是shutdown() 方法還是shutdownNow() 方法，都是無法關(guān)閉線程池的。

總結(jié)

ThreadPoolExecutor的使用總結(jié)如下。

通過ThreadPoolExecutor的execute() 方法能夠執(zhí)行Runnable任務(wù)；
通過ThreadPoolExecutor的submit() 方法能夠執(zhí)行Runnable任務(wù)和Callable任務(wù)，并且能夠獲取異步的執(zhí)行結(jié)果；
ThreadPoolExecutor的submit() 方法會返回一個Future對象（實(shí)際就是FutureTask），如果任務(wù)執(zhí)行過程中發(fā)生了異常且未捕獲，那么可以通過Future的get() 方法感知到異常；
ThreadPoolExecutor的submit() 方法無論是提交Runnable任務(wù)還是Callable任務(wù)，都是將任務(wù)封裝成了RunnableFuture接口的子類FutureTask，然后調(diào)用ThreadPoolExecutor的execute() 方法來執(zhí)行FutureTask；
關(guān)閉線程池時，如果運(yùn)行的任務(wù)可以在有限時間內(nèi)運(yùn)行完畢，那么可以使用shutdown() 方法來關(guān)閉線程池，這能夠保證在關(guān)閉線程池時，正在運(yùn)行的任務(wù)會順利運(yùn)行完畢；
關(guān)閉線程池時，如果運(yùn)行的任務(wù)永遠(yuǎn)不會結(jié)束但是響應(yīng)中斷，那么可以使用shutdownNow() 方法來關(guān)閉線程池，這種方式不保證任務(wù)順利運(yùn)行完畢；
如果任務(wù)永遠(yuǎn)不會結(jié)束且不響應(yīng)中斷，那么無論是shutdown() 方法還是shutdownNow() 方法，都無法關(guān)閉線程池。