2. 程式人生 > >Java併發程式設計之執行緒池的使用

Java併發程式設計之執行緒池的使用

阿新 • • 發佈：2019-11-01

1. 為什麼要使用多執行緒？

隨著科技的進步，現在的電腦及伺服器的處理器數量都比較多，以後可能會越來越多，比如我的工作電腦的處理器有8個，怎麼檢視呢？

計算機右鍵--屬性--裝置管理器，開啟屬性視窗，然後點選“裝置管理器”，在“處理器”下可看到所有的處理器：

也可以通過以下Java程式碼獲取到處理器的個數：

System.out.println("CPU個數:" + Runtime.getRuntime().availableProcessors());

執行結果如下所示：

CPU個數:8

既然處理器的個數增加了，如果還使用傳統的序列程式設計，就有點浪費資源了，因此，為了提高資源利用率，讓各個處理器都忙碌起來，就需要引入併發程式設計，要引入併發程式設計，就引入了多執行緒。

可以說，使用多執行緒的最直接目的就是為了提高資源利用率，資源的利用率提高了，系統的吞吐率也就相應提高了。

2. 為什麼要使用執行緒池？

在一定的範圍內，增加執行緒可以提高應用程式的吞吐率，但執行緒並不是越多越好（因為執行緒的建立與銷燬都需要很大的開銷），如果超過了某個範圍，不僅會降低應用程式的執行速度，嚴重的話，應用程式甚至會崩潰，以至於不得不重啟應用程式。

為了避免這種問題，就需要對應用程式可以建立的執行緒數量進行限制，確保線上程數量達到限制時，程式也不會耗盡資源，執行緒池就是為了解決這種問題而出現的。

執行緒池：管理一組工作執行緒的資源池。

執行緒池與工作佇列密切相關，工作佇列中儲存了所有等待執行的任務。

工作者執行緒的任務就是從工作佇列中獲取一個任務，執行任務，然後返回執行緒池並等待下一個任務。

使用執行緒池可以帶來以下好處：

  1. 通過重用現有的執行緒而不是建立新執行緒，可以在處理多個任務時減少線上程建立與銷燬過程中產生的巨大開銷。
  2. 當任務到達時，工作執行緒通常已經存在，因此不會由於等待建立執行緒而延遲任務的執行，從而提高了響應性。
  3. 可以通過調整執行緒池的大小，建立足夠多的執行緒使處理器保持忙碌狀態，同時還可以防止過多執行緒相互競爭資源而使應用程式耗盡記憶體或崩潰。

3. 建立執行緒池

3.1 使用Executors靜態方法建立（不推薦）

Executors類提供了以下4個靜態方法來快速的建立執行緒池：

  1. newFixedThreadPool
  2. newCachedThreadPool
  3. newSingleThreadExecutor
  4. newScheduledThreadPool

首先看下newFixedThreadPool()方法的使用方式：

ExecutorService threadPool = Executors.newFixedThreadPool(10);

它的原始碼為：

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

說明：newFixedThreadPool將建立一個固定長度的執行緒池，每當提交一個任務時就建立一個執行緒，直到達到執行緒池的最大數量，這時執行緒池的規模將不再變化（如果某個執行緒由於發生了未預期的Exception而結束，那麼執行緒池會補充一個新的執行緒）。

然後看下newCachedThreadPool()方法的使用方式：

ExecutorService threadPool = Executors.newCachedThreadPool();

它的原始碼為：

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

說明：newCachedThreadPool將建立一個可快取的執行緒池，如果執行緒池的規模超過了處理需求時，那麼將回收空閒的執行緒，而當需求增加時，則新增新的執行緒，執行緒池的最大規模為Integer.MAX_VALUE。

然後看下newSingleThreadExecutor()方法的使用方式：

ExecutorService threadPool = Executors.newSingleThreadExecutor();

它的原始碼為：

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

說明：newSingleThreadExecutor是一個單執行緒的Executor，它建立單個工作者執行緒來執行任務，如果這個執行緒異常結束，就建立一個新的執行緒來替代。

newSingleThreadExecutor可以確保依照任務在佇列中的順序來序列執行。

最後看下newScheduledThreadPool()方法的使用方式：

ExecutorService threadPool = Executors.newScheduledThreadPool(10);

它的原始碼為：

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

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

super指向如下程式碼：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

說明：newScheduledThreadPool將建立一個固定長度的執行緒池，而且以延遲或者定時的方式來執行任務，類似於Timer。

可以發現，以上4種方式最終都指向了ThreadPoolExecutor的以下建構函式，只是很多引數沒讓你指定，傳遞了預設值而已：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    // 省略具體的程式碼
}

雖然使用這4個方法可以快速的建立執行緒池，但還是不推薦使用，第一，很多引數都設定了預設值，不便於你理解各個引數的具體含義，第二，引數的預設值可能會造成一定的問題，最好是由使用者根據自己的需求自行指定。

那麼這7個引數分別代表什麼含義呢？請接著往下看。

3.2 使用ThreadPoolExecutor建構函式建立（推薦）

ThreadPoolExecutor共有以下4個建構函式，推薦使用這種方式來建立執行緒池：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         threadFactory, defaultHandler);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          RejectedExecutionHandler handler) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), handler);
}

以上3個也都指向引數最全的第4個建構函式：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    // 省略具體的程式碼
}

以下為各個引數的講解：

  • corePoolSize：核心執行緒數。

  • maximumPoolSize：最大執行緒數。

    最大執行緒數=核心執行緒數+非核心執行緒數。

  • keepAliveTime：非核心執行緒閒置超時時間。

    一個非核心執行緒，如果不幹活(閒置狀態)的時長超過這個引數所設定的時長，就會被銷燬掉，如果設定了allowCoreThreadTimeOut = true，則會作用於核心執行緒。

  • unit：引數keepAliveTime的時間單位，如秒、分、小時。

  • workQueue：工作佇列，即要執行的任務佇列，裡面儲存等待執行的任務。

    這裡的阻塞佇列可選擇的有：LinkedBlockingQueue、ArrayBlockingQueue、SynchronousQueue、DelayedWorkQueue。

    newFixedThreadPool()方法預設使用的LinkedBlockingQueue，

    newCachedThreadPool()方法預設使用的SynchronousQueue，

    newSingleThreadExecutor()方法預設使用的LinkedBlockingQueue，

    newScheduledThreadPool()方法預設使用的DelayedWorkQueue。

  • threadFactory：執行緒工廠，用來用來建立執行緒。

  • handler：飽和策略/拒絕處理任務時的策略。

    當workQueue已滿，並且執行緒池的執行緒數已達到maximumPoolSize，此時新提交的任務會交由RejectedExecutionHandler handler處理，主要有以下4種策略：

    AbortPolicy：中止策略，拋棄任務並丟擲未檢查的RejectedExecutionException，這也是預設的飽和策略。

    DiscardPolicy：拋棄策略，直接拋棄任務，但不丟擲異常。

    DiscardOldestPolicy：拋棄最舊的策略，拋棄下一個將被執行的任務，然後嘗試重新提交新的任務。

    CallerRunsPolicy：呼叫者執行策略，將任務回退到呼叫者，在呼叫者所在的執行緒執行該任務。

4. 執行緒池的執行原理

可以通過下面2張圖來理解執行緒池的執行原理：

1)如果執行緒池中的執行緒小於corePoolSize，則建立新執行緒來處理任務，這時建立的執行緒為核心執行緒。

2)如果執行緒中的執行緒等於或者大於corePoolSize，則將任務放到工作佇列中，即上圖中的BlockingQueue。

3)如果工作佇列已滿，無法將任務加入到BlockingQueue，則建立新的執行緒來處理任務，這時建立的執行緒為非核心執行緒，非核心執行緒在空閒一段時間後會被回收銷燬掉（keepAliveTime和unit就是用來定義這個空閒的時間是多少）。

4)如果建立新執行緒導致執行緒池中的執行緒數超過了maximumPoolSize，任務將被拒絕，並呼叫RejectedExecutionHandler.rejectedExecution()方法。

5. ThreadPoolExecutor示例

新建如下示例程式碼，建立1個corePoolSize為2，maximumPoolSize為3的執行緒池：

import java.util.concurrent.*;

public class ThreadPoolExecutorTest {
    public static void main(String[] args) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 3, 60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1));

        threadPoolExecutor.execute(() -> {
            try {
                Thread.sleep(3 * 1000);
                System.out.println("任務1執行執行緒：" + Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        threadPoolExecutor.execute(() -> {
            System.out.println("任務2執行執行緒：" + Thread.currentThread().getName());
        });
    }
}

執行結果為：

任務2執行執行緒：pool-1-thread-2

任務1執行執行緒：pool-1-thread-1

可以看出，因為執行緒池中的執行緒數小於corePoolSize，執行緒池建立了2個核心執行緒來分別執行任務1和任務2。

修改程式碼為如下所示，開啟3個任務：

import java.util.concurrent.*;

public class ThreadPoolExecutorTest {
    public static void main(String[] args) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 3, 60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1));

        threadPoolExecutor.execute(() -> {
            try {
                Thread.sleep(3 * 1000);
                System.out.println("任務1執行執行緒：" + Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        threadPoolExecutor.execute(() -> {
            try {
                Thread.sleep(5 * 1000);
                System.out.println("任務2執行執行緒：" + Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        threadPoolExecutor.execute(() -> System.out.println("任務3執行執行緒：" + Thread.currentThread().getName()));
    }
}

執行結果為：

任務1執行執行緒：pool-1-thread-1

任務3執行執行緒：pool-1-thread-1

任務2執行執行緒：pool-1-thread-2

可以看出，執行任務3時並沒有新建執行緒，而是先放入了工作佇列，最後由執行緒1執行完成。

在上面的程式碼中新增個任務4：

threadPoolExecutor.execute(() -> System.out.println("任務4執行執行緒：" + Thread.currentThread().getName()));

此時執行結果為：

任務4執行執行緒：pool-1-thread-3

任務3執行執行緒：pool-1-thread-3

任務1執行執行緒：pool-1-thread-1

任務2執行執行緒：pool-1-thread-2

可以看出，任務3是先放入了工作佇列，任務4放不到工作佇列（空間已滿），所以建立了第3個執行緒來執行，執行完畢後從佇列裡獲取到任務3執行，任務1和任務2分別由執行緒1和執行緒2執行。

修改下任務4的程式碼，並新增任務5：

threadPoolExecutor.execute(() -> {
    try {
        Thread.sleep(2 * 1000);
        System.out.println("任務4執行執行緒：" + Thread.currentThread().getName());
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
});

threadPoolExecutor.execute(() -> System.out.println("任務5執行執行緒：" + Thread.currentThread().getName()));

此時執行結果為：

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task ThreadPoolExecutorTest$$Lambda$5/935044096@179d3b25 rejected from java.util.concurrent.ThreadPoolExecutor@254989ff[Running, pool size = 3, active threads = 3, queued tasks = 1, completed tasks = 0]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
at ThreadPoolExecutorTest.main(ThreadPoolExecutorTest.java:37)
任務4執行執行緒：pool-1-thread-3

任務3執行執行緒：pool-1-thread-3

任務1執行執行緒：pool-1-thread-1

任務2執行執行緒：pool-1-thread-2

可以看出，當提交任務5時，由於工作佇列已滿， 且執行緒池中的執行緒數已經為3，所以該任務被拋棄並丟擲了java.util.concurrent.RejectedExecutionException異常。

如果你看到了這裡，是否會好奇引數maximumPoolSize設定為多少合適呢？

這個問題，我們下次講解，歡迎持續關注，哈哈！

6. 原始碼及參考

Brian Goetz《Java併發程式設計實戰》

怎麼檢視處理器（cpu）的核數

ThreadPoolExecutor使用方法

Java執行緒池-ThreadPoolExecutor原理分析與實戰

深入理解 Java 多執行緒核心知識：跳槽面試必備

網際網路大廠Java面試題：使用無界佇列的執行緒池會導致記憶體飆升嗎？【石杉的架構筆記】

最後，歡迎關注我的微信公眾號：「申城異鄉人」，所有部落格會同步更新

相關推薦

Java併發程式設計之執行緒池(三)

一.介紹 Java通過Executors提供四種執行緒池，分別為： (1)newCachedThreadPool:建立一個可快取執行緒池，如果執行緒池長度超過處理需要，可靈活回收空閒執行緒，若無可回收，則新建執行緒。 (2)newFixedThreadPool: 建立一個定長執行緒池，可控制

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

前言 本文介紹幾種java常用的執行緒池如：FixedThreadPool，ScheduledThreadPool，CachedThreadPool等執行緒池，並分析介紹Executor框架，做到“知其然”：會用執行緒池，正確使用執行緒池。並且“知其所以然”：

Java併發程式設計之執行緒池、Callable和Future使用

知識儲備 收藏幾篇好文章： 目錄結構 Callable和Future使用 執行緒池使用 Callable和Future使用 多執行緒實現方式很多，分為兩類：1、沒有返回值的；2、有返回值的。 針對“沒有返回值的”這類可以參

Java併發程式設計之執行緒池的使用

1. 為什麼要使用多執行緒？ 隨著科技的進步，現在的電腦及伺服器的處理器數量都比較多，以後可能會越來越多，比如我的工作電腦的處理器有8個，怎麼檢視呢？ 計算機右鍵--屬性--裝置管理器，開啟屬性視窗，然後點選“裝置管理器”，在“處理器”下可看到所有的處理器： 也可以通過以下Java程式碼獲取到處理器的

java併發程式設計之執行緒,執行緒池,ansync執行緒池原始碼解析

前言 java開源長了, 程式碼久了,網上對於執行緒那是眾說紛紜,一直縈繞我心頭的,jdk執行緒池好還是spring執行緒池好

Python併發程式設計之執行緒池/程序池

Python併發程式設計之執行緒池/程序池 2017/01/18 · 基礎知識 · 2 評論 · 併發, 執行緒池, 程序池 原文出處： ZiWenXie    引言 Pyt

Java併發程式設計之執行緒生命週期、守護執行緒、優先順序和join、sleep、yield

Java併發程式設計中，其中一個難點是對執行緒生命週期的理解，和多種執行緒控制方法、執行緒溝通方法的靈活運用。這些方法和概念之間彼此聯絡緊密，共同構成了Java併發程式設計基石之一。 Java執行緒的生命週期 Java執行緒類定義了New、Runnable、Running Man、Blocked和Dead

Java併發程式設計之執行緒安全、執行緒通訊

Java多執行緒開發中最重要的一點就是執行緒安全的實現了。所謂Java執行緒安全，可以簡單理解為當多個執行緒訪問同一個共享資源時產生的資料不一致問題。為此，Java提供了一系列方法來解決執行緒安全問題。 synchronized synchronized用於同步多執行緒對共享資源的訪問，在實現中分為同步程

java併發程式設計一一執行緒池原理分析（三）

合理的設定執行緒池的大小 接著上一篇探討執行緒留下的尾巴。如果合理的設定執行緒池的大小。 要想合理的配置執行緒池的大小、首先得分析任務的特性，可以從以下幾個角度分析： 1、任務的性質：CPU密集型任務、IO密集型任務、混合型任務等； 2、任務的優先順序：高、中、低； 3、任務的執行時

java併發程式設計一一執行緒池原理分析(二)

2、執行緒池 1、什麼是執行緒池 Java中的執行緒池是運用場景最多的併發框架，幾乎所有需要非同步或併發執行任務的程式都可以使用執行緒池。 在開發工程中，合理的使用執行緒池能夠帶來3個好處。 第一：降低資源的消耗，通過重複利用已建立的執行緒降低執行緒建立和銷燬造成的消耗 第二：提

java併發程式設計一一執行緒池原理分析(一)

1、併發包 1、CountDownLatch（計數器） CountDownLatch 類位於 java.util.concurrent 包下，利用它可以實現類似於計數器的功能。 比如有一個任務A，它要等待其他4個任務執行完成之後才能執行，此時就可以利用CountDownLatch

java併發程式設計之執行緒的基本概念

本文為學習筆記。源自學習微信公眾號“我們都是小青蛙”。 本篇文章將記錄如何使用java中的執行緒。 main執行緒 main方法是程式入口，我們對已經編譯好的class檔案呼叫java命令時就可以執行一個java程式。這個過程中，其實系統自動為我們建立了一個程序

Java併發（二十一）：執行緒池實現原理 Java併發（十八）：阻塞佇列BlockingQueue Java併發（十八）：阻塞佇列BlockingQueue Java併發程式設計：執行緒池的使用

一、總覽 執行緒池類ThreadPoolExecutor的相關類需要先了解：  （圖片來自：https://javadoop.com/post/java-thread-pool#%E6%80%BB%E8%A7%88） Executor：位於最頂層，只有一個 execute(Runnab

Java併發程式設計：執行緒池的使用

如果併發的執行緒數量很多，並且每個執行緒都是執行一個時間很短的任務就結束了，這樣頻繁建立執行緒就會大大降低系統的效率，因為頻繁建立執行緒和銷燬執行緒需要時間。 　　那麼有沒有一種辦法使得執行緒可以複用，就是執行完一個任務，並不被銷燬，而是可以繼續執行其他的任務？ 　　在J

Java併發程式設計：執行緒池的使用（轉載）

轉載自：https://www.cnblogs.com/dolphin0520/p/3932921.html Java併發程式設計：執行緒池的使用 　　在前面的文章中，我們使用執行緒的時候就去建立一個執行緒，這樣實現起來非常簡便，但是就會有一個問題： 　　如果併發的執行緒數量很多，並且每個執行緒都是執行

JAVA併發程式設計：執行緒池 ThreadPoolExecutor

生活 前期追深度，否則會華而不實，後期追廣度，否則會坐井觀天； 前言 在前面，我們已經對Thread有了比較深入的瞭解，並且已經學會了通過new Thread()來建立一個執行緒，並通過start方法來啟動一個執行緒，這種方法非常簡單，同樣也存在弊端： 1、每次通過new Thr

JAVA併發程式設計：執行緒池Executors

Java中對執行緒池提供了很好的支援，有了執行緒池，我們就不需要自已再去建立執行緒。如果併發的執行緒數量很多，並且每個執行緒都是執行一個時間很短的任務就結束了，頻繁建立執行緒就會大大降低系統的效率，因為頻繁建立執行緒和銷燬執行緒需要時間。JAVA的執行緒池中的執行緒可以在執

【java併發程式設計】執行緒池原理分析及ThreadPoolExecutor原始碼實現

執行緒池簡介：  多執行緒技術主要解決處理器單元內多個執行緒執行的問題，它可以顯著減少處理器單元的閒置時間，增加處理器單元的吞吐能力。         假設一個伺服器完成一項任務所需時間為：T1 建立執行緒時間，T2 線上程中執行任務的時間，T3 銷燬執行緒時間。    

Java併發程式設計之執行緒管理（基本執行緒同步3）

 (上一節的繼續) 2.2 使用Lock機制         Java提供了另外一種機制來同步程式碼塊。它是比synchrozied關鍵字更為強大且彈性的機制。它是基於鎖介面和實現了這個介面的類

JAVA 併發程式設計-基於執行緒池設計的ScheduledExecutor（八）

    上篇部落格《JAVA 併發程式設計-執行緒池（七）》中曾介紹到newScheduledThreadPool(intcorePoolSize)，建立corePoolSize大小的執行緒池。此執行緒池支援定時以及週期性執行任務的需求。    接下來我們一起來分析一下Jav