1. 程式人生 > >併發程式設計之執行緒池ThreadPoolExecutor

併發程式設計之執行緒池ThreadPoolExecutor

前言

在我們平時自己寫執行緒的測試demo時,一般都是用new Thread的方式來建立執行緒。但是,我們知道建立執行緒物件,就會在記憶體中開闢空間,而執行緒中的任務執行完畢之後,就會銷燬。

單個執行緒的話還好,如果執行緒的併發數量上來之後,就會頻繁的建立和銷燬物件。這樣,勢必會消耗大量的系統資源,進而影響執行效率。

所以,執行緒池就應運而生。

執行緒池ThreadPoolExecutor

可以通過idea先看下執行緒池的類圖,瞭解一下它的繼承關係和大概結構。

它繼承自AbstractExecutorService類,這是一個抽象類,不過裡邊的方法都是已經實現好的。然後這個類實現了ExecutorService介面,裡邊聲明瞭各種方法,包括關閉執行緒池,以及執行緒池是否已經終止等。此介面繼承自父介面Executor,裡邊只聲明瞭一個execute方法。

執行緒池就是為了解決單個執行緒頻繁的建立和銷燬帶來的效能開銷。同時,可以幫我們自動管理執行緒。並且不需要每次執行新任務都去建立新的執行緒,而是重複利用已有的執行緒,大大提高任務執行效率。

我們開啟 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.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
 }

1)corePoolSize

代表核心執行緒數。每當新的任務提交過來的時候,執行緒池就會建立一個核心執行緒來執行這個任務,即使已經有其他的核心執行緒處於空閒狀態。 而當需要執行的任務數大於核心執行緒數時,將不再建立新的核心執行緒。

其實,我們可以看下JDK提供的官方註釋說明。even if they are idle,就照應上邊的加粗字型。

此外,最後一句話說,除非allowCoreThreadTimeOut 這個引數被設定了值。


什麼意思呢,可以去看下這個引數預設值是false,代表當核心執行緒在空閒狀態時,即沒有任務在執行,就會一直存活,不會銷燬。而設定為true之後,就會有執行緒存活時間,即假如設定存活時間60秒,則60秒之後,如果沒有新的可執行任務,則核心執行緒也會自動銷燬。

2)maximumPoolSize

執行緒所允許的最大數量。即,當阻塞佇列已滿的時候,並且已經建立的執行緒數小於最大執行緒數,則會建立新的執行緒去執行任務。所以,這個引數只有在阻塞佇列滿的情況下才有意義。因此,對於無界佇列,這個引數將會失去效果。

3)keepAliveTime

代表執行緒空閒後,保持存活的時間。也就是說,超過一定的時間沒有任務執行,執行緒就會自動銷燬。

注意,這個引數,是針對大於核心執行緒數,小於最大執行緒數的那部分非核心執行緒來說的。如果是任務數量特別多的情況下,可以適當增加這個引數值的大小。以保證,在下個任務到來之前,此執行緒不會立即銷燬,從而避免執行緒的重新建立。

4)unit

這個是描述存活時間的時間單位。可以使用TimeUnit裡邊的列舉值。

5)workQueue

代表阻塞佇列,儲存所有等待執行的任務。

6)threadFactory

代表用來建立執行緒的工廠。可以自定義一個工廠,傳參進來。如果不指定的話,就會使用預設工廠(Executors類裡邊的 DefaultThreadFactory)。

可以看到,會給每個執行緒的名字指定一個有規律的字首。並且每個執行緒都設定相同的優先順序(優先順序總共有三個,1、5、10)。優先順序可以理解為,優先順序高的執行緒被執行的概率會更高,但是不代表優先順序高的執行緒一定會先執行。

7)handler

這個引數代表,拒絕策略。當阻塞佇列和執行緒池都滿了,即達到了最大執行緒數,會用什麼策略來處理。一共有四種策略可供選擇,分別對應四個內部類。

  • AbortPolicy:直接拒絕,並丟擲異常,這也是預設的策略。
  • CallerRunsPolicy:直接讓呼叫execute方法的執行緒去執行此任務。
  • DiscardOldestPolicy:丟棄最老的未處理的任務,然後重新嘗試執行當前的新任務。
  • DiscardPolicy:直接丟棄當前任務,但是不拋異常。

總結一下執行緒池的執行過程。

  1. 當執行緒數量未達到corePoolSize的時候,就會建立新的執行緒來執行任務。
  2. 當核心執行緒數已滿,就會把任務放到阻塞佇列。
  3. 當佇列已滿,並且未達到最大執行緒數,就會新建非核心執行緒來執行任務。
  4. 當佇列已滿,並且達到了最大執行緒數,則選擇一種拒絕策略來執行。

執行緒池常用的一些方法

我們一般用 execute 方法來提交任務給執行緒池。當執行緒需要返回值時,可以使用submit 方法。

shutdown方法用來關閉執行緒池。注意,此時不再接受新提交的任務,但是,會繼續處理正在執行的任務和阻塞佇列裡邊的任務。

shutdownNow也會關閉執行緒池。但是,它不再接受新任務,並且會嘗試終止正在執行的任務。

用Executors建立執行緒池

瞭解了執行緒池工作流程之後,那麼我們怎樣去建立它呢。

Executors類提供了四種常用的方法。可以發現它們最終都呼叫了執行緒池的構造方法。都有兩種建立方式,其中一種可以傳自定義的執行緒工廠。此處,只貼出不帶工廠的方法便於理解。


①newFixedThreadPool

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

建立一個固定大小的執行緒池。核心執行緒數和最大執行緒數相等。當執行緒數量達到核心執行緒數時,新任務就會放到阻塞佇列裡邊等待執行。

②newSingleThreadExecutor

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

建立一個核心執行緒數和最大執行緒數都是1的執行緒池。即執行緒池中只會存在一個正在執行的執行緒,若執行緒空閒則執行,否則把任務放到阻塞佇列。

③ newCachedThreadPool

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

建立一個可根據實際情況調整執行緒個數的執行緒池。這句話,可以理解為,有多少任務同時進來,就會建立同等數量的執行緒去執行任務。當然,這是線上程數不能超過Integer最大值的前提下。

當再來一個新任務時,若有空閒執行緒則執行任務。否則,等執行緒空閒60秒之後,就會自動回收。

當沒有新任務,就不會建立新的執行緒。

④newScheduledThreadPool

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
}
    
public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
}

建立一個可指定核心執行緒數的執行緒池。這個執行緒池可以執行週期性的任務。
如果本文對你有用,歡迎點贊,評論,轉發。

學習是枯燥的,也是有趣的。我是「煙雨星空」,歡迎關注,可第一時間接收文章推送