Java多線程之徹底搞懂線程池

更新時間：2023年03月27日 10:34:46 作者：孫強 Jimmy

熟悉 Java 多線程編程的同學(xué)都知道，當我們線程創(chuàng)建過多時，容易引發(fā)內(nèi)存溢出，因此我們就有必要使用線程池的技術(shù)了。想要詳細了解的同學(xué)可以參考閱讀本文

1 線程池的優(yōu)勢

總體來說，線程池有如下的優(yōu)勢：

（1）降低資源消耗。通過重復(fù)利用已創(chuàng)建的線程降低線程創(chuàng)建和銷毀造成的消耗。

（2）提高響應(yīng)速度。當任務(wù)到達時，任務(wù)可以不需要等到線程創(chuàng)建就能立即執(zhí)行。

（3）提高線程的可管理性。線程是稀缺資源，如果無限制的創(chuàng)建，不僅會消耗系統(tǒng)資源，還會降低系統(tǒng)的穩(wěn)定性，使用線程池可以進行統(tǒng)一的分配，調(diào)優(yōu)和監(jiān)控。

2 線程池的使用

線程池的真正實現(xiàn)類是 ThreadPoolExecutor，其構(gòu)造方法有如下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);
}
 
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;
}

可以看到，其需要如下幾個參數(shù)：

corePoolSize（必需）：核心線程數(shù)。默認情況下，核心線程會一直存活，但是當將 allowCoreThreadTimeout 設(shè)置為 true 時，核心線程也會超時回收。
maximumPoolSize（必需）：線程池所能容納的最大線程數(shù)。當活躍線程數(shù)達到該數(shù)值后，后續(xù)的新任務(wù)將會阻塞。
keepAliveTime（必需）：線程閑置超時時長。如果超過該時長，非核心線程就會被回收。如果將 allowCoreThreadTimeout 設(shè)置為 true 時，核心線程也會超時回收。
unit（必需）：指定 keepAliveTime 參數(shù)的時間單位。常用的有：TimeUnit.MILLISECONDS（毫秒）、TimeUnit.SECONDS（秒）、TimeUnit.MINUTES（分）。
workQueue（必需）：任務(wù)隊列。通過線程池的 execute() 方法提交的 Runnable 對象將存儲在該參數(shù)中。其采用阻塞隊列實現(xiàn)。
threadFactory（可選）：線程工廠。用于指定為線程池創(chuàng)建新線程的方式。

線程池的使用流程如下：

// 創(chuàng)建線程池
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(CORE_POOL_SIZE,
                                             MAXIMUM_POOL_SIZE,
                                             KEEP_ALIVE,
                                             TimeUnit.SECONDS,
                                             sPoolWorkQueue,
                                             sThreadFactory);
// 向線程池提交任務(wù)
threadPool.execute(new Runnable() {
    @Override
    public void run() {
        ... // 線程執(zhí)行的任務(wù)
    }
});
// 關(guān)閉線程池
threadPool.shutdown(); // 設(shè)置線程池的狀態(tài)為SHUTDOWN，然后中斷所有沒有正在執(zhí)行任務(wù)的線程
threadPool.shutdownNow(); // 設(shè)置線程池的狀態(tài)為 STOP，然后嘗試停止所有的正在執(zhí)行或暫停任務(wù)的線程，并返回等待執(zhí)行任務(wù)的列表

3 線程池的工作原理

下面來描述一下線程池工作的原理，同時對上面的參數(shù)有一個更深的了解。其工作原理流程圖如下：

通過上圖，相信大家已經(jīng)對所有參數(shù)有個了解了。下面再對任務(wù)隊列、線程工廠和拒絕策略做更多的說明。

4 線程池的參數(shù)

4.1 任務(wù)隊列（workQueue）

任務(wù)隊列是基于阻塞隊列實現(xiàn)的，即采用生產(chǎn)者消費者模式，在 Java 中需要實現(xiàn) BlockingQueue 接口。但 Java 已經(jīng)為我們提供了 7 種阻塞隊列的實現(xiàn)：

ArrayBlockingQueue：一個由數(shù)組結(jié)構(gòu)組成的有界阻塞隊列（數(shù)組結(jié)構(gòu)可配合指針實現(xiàn)一個環(huán)形隊列）。
LinkedBlockingQueue：一個由鏈表結(jié)構(gòu)組成的有界阻塞隊列，在未指明容量時，容量默認為 Integer.MAX_VALUE。
PriorityBlockingQueue：一個支持優(yōu)先級排序的無界阻塞隊列，對元素沒有要求，可以實現(xiàn) Comparable 接口也可以提供 Comparator 來對隊列中的元素進行比較。跟時間沒有任何關(guān)系，僅僅是按照優(yōu)先級取任務(wù)。
DelayQueue：類似于PriorityBlockingQueue，是二叉堆實現(xiàn)的無界優(yōu)先級阻塞隊列。要求元素都實現(xiàn) Delayed 接口，通過執(zhí)行時延從隊列中提取任務(wù)，時間沒到任務(wù)取不出來。
SynchronousQueue：一個不存儲元素的阻塞隊列，消費者線程調(diào)用 take() 方法的時候就會發(fā)生阻塞，直到有一個生產(chǎn)者線程生產(chǎn)了一個元素，消費者線程就可以拿到這個元素并返回；生產(chǎn)者線程調(diào)用 put() 方法的時候也會發(fā)生阻塞，直到有一個消費者線程消費了一個元素，生產(chǎn)者才會返回。
LinkedBlockingDeque：使用雙向隊列實現(xiàn)的有界雙端阻塞隊列。雙端意味著可以像普通隊列一樣 FIFO（先進先出），也可以像棧一樣 FILO（先進后出）。
LinkedTransferQueue：它是ConcurrentLinkedQueue、LinkedBlockingQueue 和 SynchronousQueue 的結(jié)合體，但是把它用在 ThreadPoolExecutor 中，和 LinkedBlockingQueue 行為一致，但是是無界的阻塞隊列。

注意有界隊列和無界隊列的區(qū)別：如果使用有界隊列，當隊列飽和時并超過最大線程數(shù)時就會執(zhí)行拒絕策略；而如果使用無界隊列，因為任務(wù)隊列永遠都可以添加任務(wù)，所以設(shè)置 maximumPoolSize 沒有任何意義。

4.2 線程工廠（threadFactory）

線程工廠指定創(chuàng)建線程的方式，需要實現(xiàn) ThreadFactory 接口，并實現(xiàn) newThread(Runnable r) 方法。該參數(shù)可以不用指定，Executors 框架已經(jīng)為我們實現(xiàn)了一個默認的線程工廠：

/**
 * The default thread factory.
 */
private static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;
 
    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }
 
    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

4.3 拒絕策略（handler）

當線程池的線程數(shù)達到最大線程數(shù)時，需要執(zhí)行拒絕策略。拒絕策略需要實現(xiàn) RejectedExecutionHandler 接口，并實現(xiàn) rejectedExecution(Runnable r, ThreadPoolExecutor executor) 方法。不過 Executors 框架已經(jīng)為我們實現(xiàn)了 4 種拒絕策略：

AbortPolicy（默認）：丟棄任務(wù)并拋出 RejectedExecutionException 異常。
CallerRunsPolicy：由調(diào)用線程處理該任務(wù)。
DiscardPolicy：丟棄任務(wù)，但是不拋出異常?？梢耘浜线@種模式進行自定義的處理方式。
DiscardOldestPolicy：丟棄隊列最早的未處理任務(wù)，然后重新嘗試執(zhí)行任務(wù)。

5 功能線程池

嫌上面使用線程池的方法太麻煩？其實Executors已經(jīng)為我們封裝好了 4 種常見的功能線程池，如下：

定長線程池（FixedThreadPool）
定時線程池（ScheduledThreadPool ）
可緩存線程池（CachedThreadPool）
單線程化線程池（SingleThreadExecutor）

5.1 定長線程池（FixedThreadPool）

創(chuàng)建方法的源碼：

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

特點：只有核心線程，線程數(shù)量固定，執(zhí)行完立即回收，任務(wù)隊列為鏈表結(jié)構(gòu)的有界隊列。
應(yīng)用場景：控制線程最大并發(fā)數(shù)。

使用示例：

// 1. 創(chuàng)建定長線程池對象 & 設(shè)置線程池線程數(shù)量固定為3
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
// 2. 創(chuàng)建好Runnable類線程對象 & 需執(zhí)行的任務(wù)
Runnable task =new Runnable(){
  public void run() {
     System.out.println("執(zhí)行任務(wù)啦");
  }
};
// 3. 向線程池提交任務(wù)
fixedThreadPool.execute(task);

5.2 定時線程池（ScheduledThreadPool ）

創(chuàng)建方法的源碼：

private static final long DEFAULT_KEEPALIVE_MILLIS = 10L;
 
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue());
}
 
public static ScheduledExecutorService newScheduledThreadPool(
        int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue(), threadFactory);
}

特點：核心線程數(shù)量固定，非核心線程數(shù)量無限，執(zhí)行完閑置 10ms 后回收，任務(wù)隊列為延時阻塞隊列。
應(yīng)用場景：執(zhí)行定時或周期性的任務(wù)。

使用示例：

// 1. 創(chuàng)建 定時線程池對象 & 設(shè)置線程池線程數(shù)量固定為5
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
// 2. 創(chuàng)建好Runnable類線程對象 & 需執(zhí)行的任務(wù)
Runnable task =new Runnable(){
  public void run() {
     System.out.println("執(zhí)行任務(wù)啦");
  }
};
// 3. 向線程池提交任務(wù)
scheduledThreadPool.schedule(task, 1, TimeUnit.SECONDS); // 延遲1s后執(zhí)行任務(wù)
scheduledThreadPool.scheduleAtFixedRate(task,10,1000,TimeUnit.MILLISECONDS);// 延遲10ms后、每隔1000ms執(zhí)行任務(wù)

5.3 可緩存線程池（CachedThreadPool）

創(chuàng)建方法的源碼：

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

特點：無核心線程，非核心線程數(shù)量無限，執(zhí)行完閑置 60s 后回收，任務(wù)隊列為不存儲元素的阻塞隊列。
應(yīng)用場景：執(zhí)行大量、耗時少的任務(wù)。

使用示例：

// 1. 創(chuàng)建可緩存線程池對象
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
// 2. 創(chuàng)建好Runnable類線程對象 & 需執(zhí)行的任務(wù)
Runnable task =new Runnable(){
  public void run() {
     System.out.println("執(zhí)行任務(wù)啦");
  }
};
// 3. 向線程池提交任務(wù)
cachedThreadPool.execute(task);

5.4 單線程化線程池（SingleThreadExecutor）

創(chuàng)建方法的源碼：

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

特點：只有 1 個核心線程，無非核心線程，執(zhí)行完立即回收，任務(wù)隊列為鏈表結(jié)構(gòu)的有界隊列。
應(yīng)用場景：不適合并發(fā)但可能引起 IO 阻塞性及影響 UI 線程響應(yīng)的操作，如數(shù)據(jù)庫操作、文件操作等

使用示例：

// 1. 創(chuàng)建單線程化線程池
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
// 2. 創(chuàng)建好Runnable類線程對象 & 需執(zhí)行的任務(wù)
Runnable task =new Runnable(){
  public void run() {
     System.out.println("執(zhí)行任務(wù)啦");
  }
};
// 3. 向線程池提交任務(wù)
singleThreadExecutor.execute(task);

5.5 對比

6 總結(jié)

Executors 的 4 個功能線程池雖然方便，但現(xiàn)在已經(jīng)不建議使用了，而是建議直接通過使用 ThreadPoolExecutor 的方式，這樣的處理方式讓寫的同學(xué)更加明確線程池的運行規(guī)則，規(guī)避資源耗盡的風險。

其實 Executors 的 4 個功能線程有如下弊端：

FixedThreadPool 和 SingleThreadExecutor：主要問題是堆積的請求處理隊列均采用 LinkedBlockingQueue，可能會耗費非常大的內(nèi)存，甚至 OOM。
CachedThreadPool 和 ScheduledThreadPool：主要問題是線程數(shù)最大數(shù)是 Integer.MAX_VALUE，可能會創(chuàng)建數(shù)量非常多的線程，甚至 OOM。

以上就是Java多線程之徹底搞懂線程池的詳細內(nèi)容，更多關(guān)于Java多線程池的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: