概述

在處理大量任務時,重復利用線程可以提高程序執行效率,因此線程池應運而生。

• 它是一種重用線程的機制,可以有效降低內存資源消耗
• 提高響應速度。當任務到達時，任務可以不需要的等到線程創建就能立即執行
• 線程池可以幫助我們更好地管理線程的生命周期和資源使用,避免線程頻繁地創建和銷毀帶來的性能問題

同時,線程池還可以提供一些額外的功能,例如線程池的大小控制、線程池的任務隊列、線程池的拒絕策略等。線程池中通常維護一個線程隊列,線程隊列中保存著已創建的線程,當有新的任務需要執行時,線程池中的線程就可以從隊列中取出一個線程來執行任務,任務執行完畢后線程可以被放回線程隊列中,等待下一個任務的到來

核心執行流程

• 當有新任務需要線程執行時,線程池會先判斷是否有空閑的核心線程,如果有則將任務分配給其中一個空閑的核心線程執行。如果沒有空閑的核心線程或者核心線程的數量還沒達到最大值，則創建一個新的核心線程來執行任務
• 若核心線程已經達到最大值且工作隊列未滿,則將新提交的任務存儲在這個工作隊列中。如果工作隊列已滿,則會判斷線程池中的線程數是否已經達到最大值,如果沒有則創建一個新的非核心線程來執行任務
• 若工作隊列已滿且線程池中的線程都已經處于工作狀態,即核心線程和非核心線程都在執行任務,則交給飽和策略來處理這個任務。飽和策略可以決定如何處理無法處理的任務,例如拋出異常或者阻塞任務提交

線程池狀態

• RUNNING：該狀態代表能接受新任務以及處理任務(初始狀態)
• SHUTDOWN：該狀態代表不接受新任務,但處理已添加的任務(調用shutdown()時,由RUNNING->SHUTDOWN)
• STOP：該狀態時表示不接受新任務,不處理已添加任務,并會中斷正在處理中的任務(調用shutdownNow()時,由RUNNING或者SHUTDOWN→STOP)
• TIDYING：進入SHUTDOWN或者STOP狀態后,所有任務都被處理或者清理干凈后就會進入該狀態,同時會執行terminated()方法（該方法是個鉤子函數,自定義實現）
• TERMINATED：結束狀態,執行完terminated方法后由TIDYING->TERMINATED

Executor 框架

兩級調度模型?

在HotSpotVM的線程模型中,Java線程(java.lang.Thread)被一對一映射為本地操作系統線程
Java線程啟動時會創建一個本地操作系統線程；當該 Java 線程終止時，這個操作系統線程也會被回收。操作系統會調度所有線程并將它們分配給可用的CPU

• 在上層架構中,Java多線程程序通常把應用分解為若干個任務,應用程序通過Executor框架將這些任務映射為固定數量的線程
• 在下層架構中,操作系統內核將這些線程映射到硬件處理器上,下層的調度不受應用程序的控制

Executor結構

Executor框架包含的主要的類與接口如圖所示

• Executor是一個接口,它是Executor框架的基礎,它將任務的提交與任務的執行分離開來
• ThreadPoolExecutor是線程池的核心實現類,用來執行被提交的任務
• ScheduledThreadPoolExecutor是一個實現類,用來延遲之后執行任務或者定時執行任務。ScheduledThreadPoolExecutor比Timer更靈活,功能更強大
• 實現Future接口的FutureTask類,代表異步計算的結果
• Runnable接口和Callable接口的實現類,都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor執行

ThreadPoolExecutor

構造函數詳細說明

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.acc = System.getSecurityManager() == null ?null :AccessController.getContext();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;
}

參數介紹:

• corePoolSize:核心線程池個大小。一般來說任務比較耗時可以配CPU核數*2,因為這樣可以充分利用CPU,任務小并且執行很快則可以配CPU核數+1或者更小(因為線程上下文切換耗時)（獲取CPU核心數：Runtime.getRuntime().availableProcessors()）
• maximumPoolSize：最大線程池大小。當線程數>=corePoolSize,且任務隊列已滿時。線程池會創建新線程來處理任務,總線程數≤maximumPoolSize
• keepAliveTime：空閑時間,超過核心線程數的線程在到達空閑時間后會被銷毀
• TimeUnit : 時間單位
• BlockingQueue：用來暫時保存任務的隊列（阻塞隊列）
• ThreadFactory：自定義的線程工廠,默認是一個新的、非守護線程并且不包含特殊的配置信息,我們也可以自定義加入我們的調試信息,比如線程名稱、錯誤日志等
• RejectedExecutionHandler：飽和策略。當線程數=maximumPoolSize,且任務隊列已滿時,多余的任務需要采取的措施,有以下幾種（默認AbortPolicy）：
  • AbortPolicy：丟棄任務并拋出RejectedExecutionException異常
  • DiscardPolicy: 丟掉這個任務并且不會有任何異常
  • DiscardOldestPolicy：丟棄最老的。也就是說如果隊列滿了，會將最早進入隊列的任務刪掉騰出空間，再嘗試加入隊列
  • CallerRunsPolicy：主線程會自己去執行該任務，不會等待線程池中的線程去執行
  • 自定義:當然也可以自定義策略

常用ThreadPoolExecutor類型

ThreadPoolExecutor通常使用工廠類Executors來創建,包括3種ThreadPoolExecutor類型：

• FixedThreadPool：可重用固定線程數的線程池
• SingleThreadExecutor：單個線程的線程池（只有一個工作線程）
• CachedThreadPool：根據需要創建新線程的線程池

FixedThreadPool

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

• 其中corePool和maximumPoolSize?都被設置成指定的參數
• keepAliveTime設置為0L,意味著多余的空閑線程會被立即終止
• 適用于為了滿足資源管理的需求,而需要限制當前線程數量的應用場景,更適合于負載比較重的服務器

• 如果當前線程少于corePool,則創建新線程來執行當前任務
• 當運行的線程數等于corePool之后,將任務加入LinkedBlockingQueue隊列中
• 線程執行完當前任務后會循環反復從LinkedBlockingQueue獲取任務來執行

由于為LinkedBlockingQueue無界隊列(長度Integer.MAX_VALUE),所以會出現如下情景：

• maximumPoolSize和keepAliveTime參數將會無效,因為maximumPoolSize=corePool
• 不會拒絕任務,因為是無界隊列,任務不會滿

SingleThreadExecutor

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

corePool和maximumPoolSize?均被設置成了1,其他影響和運行方式都與FixedThreadPool相同

CachedThreadPool

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

• corePoolSize核心線程數為0，而maximumPoolSize為Integer.MAX_VALUE(21億多),意味著沒有空閑線程就會不斷的創建線程去執行,極端情況會耗盡CPU和內存資源;
• keepAliveTime=60s后空閑線程會被終止,所以長時間內保持空閑的情況下不會占用任何資源
• SynchronousQueue是沒有容量的阻塞隊列,每個插入操作都會等待另一個線程對應的取出操作
• 適用于執行很多的短期異步任務的小程序，或者是負載較輕的服務器

ScheduleThreadPoolExecutor

構造函數

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

ScheduledThreadPoolExecutor為了實現周期性任務對ThreadPoolExecutor做了如下修改：

• 使用DelayedWorkQueue作為任務隊列
• 獲取任務的方式不同，同樣都是隊列的take，但增加了時間的判斷
• 執行周期任務后，增加了額外的處理（需要把任務重新添加進隊列）

常用ScheduleThreadPoolExecutor類型

ScheduledThreadPoolExecutor通常使用工廠類Executors來創建,2種類型：

• ScheduledThreadPoolExecutor: 包含若干個線程 ScheduledThreadPoolExecutor

  public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
  public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory) 

  • ScheduledThreadPoolExecutor 適用于需要多個后臺線程執行周期任務，同時為了滿足資源管理的需求而需要限制后臺線程的數量的應用場景

• SingleThreadScheduledExecutor: 只包含一個線程 ScheduledThreadPoolExecutor

  public static ScheduledExecutorService newSingleThreadScheduledExecutor()
  public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory)

  • SingleThreadScheduledExecutor 適用于需要單個后臺線程執行周期任務，同時需要保證順序地執行各個任務的應用場景

FutureTask

Future接口和實現Future接口的FutureTask類用來表示異步計算的結果。當我們把Runnable接口或Callable接口的實現類提交(submit)給 ThreadPoolExecutor或ScheduledThreadPoolExecutor時,ThreadPoolExecutor或ScheduledThreadPoolExecutor會向我們返回一個FutureTask對象

Executor 實踐

ThreadPoolExecutor

package com.bierce;
import java.util.concurrent.*;
public class TestThreadPoolExecutor {public static void main(String[] args) throws ExecutionException, InterruptedException {//創建線程池ExecutorService executorService = Executors.newFixedThreadPool(5);//方式一：Runnable方式的分配10個任務提交給線程池ThreadPoolDemo threadPoolDemo = new ThreadPoolDemo();for (int i = 0; i <= 10; i++) {executorService.submit(threadPoolDemo);}//方式二：Callable方式的分配10個任務提交給線程池for (int i = 0; i <= 10; i++) {Future<Object> sum = executorService.submit(() -> {int sum1 = 0;for (int i1 = 1; i1 <= 100; i1++) {sum1 += i1;}return sum1;});System.out.println(Thread.currentThread().getName() + ":" + sum.get()); //main:5050}//關閉線程池executorService.shutdown();}
}
class ThreadPoolDemo implements Runnable{private int i = 0;@Overridepublic void run() {while (i<10){System.out.println(Thread.currentThread().getName() + ":" + i++);}}
}

ScheduleThreadPoolExecutor

package com.bierce;
import java.util.Random;
import java.util.concurrent.*;
public class TestScheduleThreadPool {public static void main(String[] args) throws ExecutionException, InterruptedException {ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);for (int i = 1; i <=5 ; i++) {//調用schedule方法執行任務Future<Integer> random = scheduledExecutorService.schedule(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {int random = new Random().nextInt(100);System.out.println(Thread.currentThread().getName() + ":" + random);return random;}},1,TimeUnit.SECONDS); //每隔一秒執行一個任務System.out.println(random.get());}scheduledExecutorService.shutdown(); //關閉線程池}
}