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

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

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

前言

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

本篇文章將從線程池簡單原理，線程池的創(chuàng)建，線程池執(zhí)行任務和關閉線程池進行使用學習。

正文

一. 線程池的簡單原理

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

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

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

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

通過ThreadPoolExecutor能夠創(chuàng)建一個線程池，ThreadPoolExecutor的構造函數(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間，線程?；顣r間單位和任務阻塞隊列，并按需指定線程工廠和飽和拒絕策略，如果不指定線程工廠和飽和拒絕策略，則ThreadPoolExecutor會使用默認的線程工廠和飽和拒絕策略。下面分別介紹這些參數(shù)的含義。

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

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

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

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

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

運行測試程序，結果如下。

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

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

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

運行測試程序，結果如下。

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

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

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

運行測試程序，結果如下。

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

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

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)建結果對象
        MyResult myResult = new MyResult();
        // 創(chuàng)建Runnable對象
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                myResult.setResult("任務執(zhí)行了");
            }
        };
        // 通過ThreadPoolExecutor的submit()方法提交Runnable
        Future&lt;MyResult&gt; resultFuture = threadPoolExecutor.submit(runnable, myResult);
        // 獲取執(zhí)行結果
        MyResult finalResult = resultFuture.get();
        // myResult和finalResult的地址實際相同
        Assert.assertEquals(myResult, finalResult);
        // 打印執(zhí)行結果
        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;
        }
    }
}

運行測試程序，結果如下。

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

四. 關閉線程池

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

public class ThreadPoolExecutorTest {
    @Test
    public void 通過shutdown關閉線程池() {
        // 創(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í)行任務
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        // 調用shutdown方法關閉線程池
        threadPoolExecutor.shutdown();
        // 等待3秒觀察現(xiàn)象
        LockSupport.parkNanos(1000 * 1000 * 1000 * 3L);
    }
}

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

public class ThreadPoolExecutorTest {
    @Test
    public void 通過shutdownNow關閉線程池() {
        // 創(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í)行任務
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        // 調用shutdown方法關閉線程池
        threadPoolExecutor.shutdownNow();
        // 等待3秒觀察現(xiàn)象
        LockSupport.parkNanos(1000 * 1000 * 1000 * 3L);
    }
}

運行測試程序，打印如下。

因為測試程序中的任務是響應中斷的，而ThreadPoolExecutor的shutdownNow() 方法會中斷所有Worker，所以執(zhí)行shutdownNow() 方法后，正在運行的任務會響應中斷并結束運行，最終線程池關閉。

假如線程池中運行著一個永遠不會結束的任務，且這個任務不響應中斷，那么無論是shutdown() 方法還是shutdownNow() 方法，都是無法關閉線程池的。

總結

ThreadPoolExecutor的使用總結如下。

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