java并發ThreadPoolExecutor如何使用

這篇文章主要介紹“java并發ThreadPoolExecutor如何使用”的相關知識，小編通過實際案例向大家展示操作過程，操作方法簡單快捷，實用性強，希望這篇“java并發ThreadPoolExecutor如何使用”文章能幫助大家解決問題。

一. 線程池的簡單原理

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

• 如果當前運行的線程數小于corePoolSzie（核心線程數），則創建新線程來執行任務（需要獲取全局鎖）；

• 如果當前運行的線程數等于或大于corePoolSzie，則將任務加入BlockingQueue（任務阻塞隊列）；

• 如果BlockingQueue已滿，則創建新的線程來執行任務（需要獲取全局鎖）；

• 如果創建新線程會使當前線程數大于maximumPoolSize（最大線程數），則拒絕任務并調用RejectedExecutionHandler的rejectedExecution() 方法。

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

二. 線程池的創建

通過ThreadPoolExecutor能夠創建一個線程池，ThreadPoolExecutor的構造函數簽名如下。

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創建線程池時，需要指定線程池的核心線程數，最大線程數，線程保活時間，線程保活時間單位和任務阻塞隊列，并按需指定線程工廠和飽和拒絕策略，如果不指定線程工廠和飽和拒絕策略，則ThreadPoolExecutor會使用默認的線程工廠和飽和拒絕策略。下面分別介紹這些參數的含義。

三. 線程池執行任務

1. 執行無返回值任務

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

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

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

2. 執行有返回值任務

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

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

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

3. 執行有返回值任務時拋出錯誤

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

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

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

4. ThreadPoolExecutor通過submit方式執行Runnable

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

public class ThreadPoolExecutorTest {
    @Test
    public void ThreadPoolExecutor通過submit的方式來提交并執行Runnable() throws Exception {
        // 創建一個線程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
        // 創建結果對象
        MyResult myResult = new MyResult();
        // 創建Runnable對象
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                myResult.setResult("任務執行了");
            }
        };
        // 通過ThreadPoolExecutor的submit()方法提交Runnable
        Future<MyResult> resultFuture = threadPoolExecutor.submit(runnable, myResult);
        // 獲取執行結果
        MyResult finalResult = resultFuture.get();
        // myResult和finalResult的地址實際相同
        Assert.assertEquals(myResult, finalResult);
        // 打印執行結果
        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() 方法來執行FutureTask。

四. 關閉線程池

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

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

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

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

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

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

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

關于“java并發ThreadPoolExecutor如何使用”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識，可以關注億速云行業資訊頻道，小編每天都會為大家更新不同的知識點。