Java高並發之線程池詳解

線程池優勢

在業務場景中, 如果一個對象創建銷毀開銷比較大, 那么此時建議池化對象進行管理.

例如線程, jdbc連接等等, 在高並發場景中, 如果可以復用之前銷毀的對象, 那么系統效率將大大提升.

另外一個好處是可以設定池化對象的上限, 例如預防創建線程數量過多導致系統崩潰的場景.

jdk中的線程池

下文主要從以下幾個角度講解:

創建線程池

我們可以通過自定義ThreadPoolExecutor或者jdk內置的Executors來創建一系列的線程池

• newFixedThreadPool: 創建固定線程數量的線程池
• newSingleThreadExecutor: 創建單一線程的池
• newCachedThreadPool: 創建線程數量自動擴容, 自動銷毀的線程池
• newScheduledThreadPool: 創建支持計划任務的線程池

 上述幾種都是通過new ThreadPoolExecutor()來實現的, 構造函數源碼如下: 

    /**
     * @param corePoolSize 池內核心線程數量, 超出數量的線程會進入阻塞隊列
     * @param maximumPoolSize 最大可創建線程數量
     * @param keepAliveTime 線程存活時間
     * @param unit 存活時間的單位
     * @param workQueue 線程溢出后的阻塞隊列
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler);
    }

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
    }

    public static ExecutorService newSingleThreadExecutor() {
        return new Executors.FinalizableDelegatedExecutorService
                (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()));
    }

    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
    }

    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS, new ScheduledThreadPoolExecutor.DelayedWorkQueue());
    }

提交任務

直接調用executorService.execute(runnable)或者submit(runnable)即可,

execute和submit的區別在於submit會返回Future來獲取任何執行的結果.

我們看下newScheduledThreadPool的使用示例.

 1 public class SchedulePoolDemo {  2  3 public static void main(String[] args){  4 ScheduledExecutorService service = Executors.newScheduledThreadPool(10);  5 // 如果前面的任務沒有完成, 調度也不會啟動  6 service.scheduleAtFixedRate(new Runnable() {  7  @Override  8 public void run() {  9 try { 10 Thread.sleep(2000); 11 // 每兩秒打印一次. 12 System.out.println(System.currentTimeMillis()/1000); 13 } catch (InterruptedException e) { 14  e.printStackTrace(); 15  } 16  } 17 }, 0, 2, TimeUnit.SECONDS); 18  } 19 }

潛在宕機風險

使用Executors來創建要注意潛在宕機風險.其返回的線程池對象的弊端如下:

• FixedThreadPool和SingleThreadPoolPool : 允許的請求隊列長度為 Integer.MAX_VALUE，可能會堆積大量的請求，從而導致 OOM.
• CachedThreadPool和ScheduledThreadPool : 允許的創建線程數量為 Integer.MAX_VALUE，可能會創建大量的線程，從而導致 OOM.

綜上所述, 在可能有大量請求的線程池場景中, 更推薦自定義ThreadPoolExecutor來創建線程池, 具體構造函數配置見下文.

線程池大小配置

一般根據任務類型進行區分, 假設CPU為N核

• CPU密集型任務需要減少線程數量, 降低線程之間切換造成的開銷, 可配置線程池大小為N + 1.
• IO密集型任務則可以加大線程數量, 可配置線程池大小為 N * 2.
• 混合型任務則可以拆分為CPU密集型與IO密集型, 獨立配置.

自定義阻塞隊列BlockingQueue

主要存放等待執行的線程, ThreadPoolExecutor中支持自定義該隊列來實現不同的排隊隊列.

• ArrayBlockingQueue：先進先出隊列，創建時指定大小, 有界；
• LinkedBlockingQueue：使用鏈表實現的先進先出隊列，默認大小為Integer.MAX_VALUE；
• SynchronousQueue：不保存提交的任務, 數據也不會緩存到隊列中, 用於生產者和消費者互等對方, 一起離開.
• PriorityBlockingQueue: 支持優先級的隊列

回調接口

線程池提供了一些回調方法, 具體使用如下所示.

        ExecutorService service = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<Runnable>()) {

            @Override
            protected void beforeExecute(Thread t, Runnable r) {
                System.out.println("准備執行任務: " + r.toString());
            }

            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                System.out.println("結束任務: " + r.toString());
            }

            @Override
            protected void terminated() {
                System.out.println("線程池退出");
            }
        };

可以在回調接口中, 對線程池的狀態進行監控, 例如任務執行的最長時間, 平均時間, 最短時間等等, 還有一些其他的屬性如下:

• taskCount：線程池需要執行的任務數量.
• completedTaskCount：線程池在運行過程中已完成的任務數量.小於或等於taskCount.
• largestPoolSize：線程池曾經創建過的最大線程數量.通過這個數據可以知道線程池是否滿過.如等於線程池的最大大小，則表示線程池曾經滿了.
• getPoolSize:線程池的線程數量.如果線程池不銷毀的話，池里的線程不會自動銷毀，所以這個大小只增不減.
• getActiveCount：獲取活動的線程數.

自定義拒絕策略

線程池滿負荷運轉后, 因為時間空間的問題, 可能需要拒絕掉部分任務的執行.

jdk提供了RejectedExecutionHandler接口, 並內置了幾種線程拒絕策略

• AbortPolicy: 直接拒絕策略, 拋出異常.
• CallerRunsPolicy: 調用者自己執行任務策略.
• DiscardOldestPolicy: 舍棄最老的未執行任務策略.

使用方式也很簡單, 直接傳參給ThreadPool

        ExecutorService service = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, 
                new SynchronousQueue<Runnable>(),
                Executors.defaultThreadFactory(),
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.out.println("reject task: " + r.toString());
                    }
                });

自定義ThreadFactory

線程工廠用於創建池里的線程. 例如在工廠中都給線程setDaemon(true), 這樣程序退出的時候, 線程自動退出.

或者統一指定線程優先級, 設置名稱等等.

class NamedThreadFactory implements ThreadFactory {
    private static final AtomicInteger threadIndex = new AtomicInteger(0);
    private final String baseName;
    private final boolean daemon;

    public NamedThreadFactory(String baseName) {
        this(baseName, true);
    }

    public NamedThreadFactory(String baseName, boolean daemon) {
        this.baseName = baseName;
        this.daemon = daemon;
    }

    public Thread newThread(Runnable runnable) {
        Thread thread = new Thread(runnable, this.baseName + "-" + threadIndex.getAndIncrement());
        thread.setDaemon(this.daemon); 18         return thread;
    }
}

關閉線程池

跟直接new Thread不一樣, 局部變量的線程池, 需要手動關閉, 不然會導致線程泄漏問題.

默認提供兩種方式關閉線程池.

• shutdown: 等所有任務, 包括阻塞隊列中的執行完, 才會終止, 但是不會接受新任務.
• shutdownNow: 立即終止線程池, 打斷正在執行的任務, 清空隊列.