一篇文章帶你深入了解Java線程池

更新時間：2021年08月02日 11:50:10 作者：Java盤魚宴

這篇文章主要介紹了Java 線程池的相關資料，文中講解非常細致，幫助大家更好的理解和學習，感興趣的朋友可以了解下，希望能給你帶來幫助

線程池模型

一般的池化模型會有兩個方法，用于獲取資源和釋放資源，就像這樣：

public interface XXPool{
    XX acquire(); 
    void release();
}

但是，工程中的線程池一般是生產者和消費者模型，線程池是消費者，任務的提交者是生產者，下面是一個簡化的線程池模型：

//簡化的線程池，僅用來說明工作原理
class MyThreadPool{
  //利用阻塞隊列實現生產者-消費者模式
  BlockingQueue<Runnable> workQueue;
  //保存內部工作線程
  List<WorkerThread> threads 
    = new ArrayList<>();
  // 構造方法
  MyThreadPool(int poolSize, 
    BlockingQueue<Runnable> workQueue){
    this.workQueue = workQueue;
    // 創(chuàng)建工作線程
    for(int idx=0; idx<poolSize; idx++){
      WorkerThread work = new WorkerThread();
      work.start();
      threads.add(work);
    }
  }
  // 提交任務
  void execute(Runnable command){
    workQueue.put(command);
  }
  // 工作線程負責消費任務，并執(zhí)行任務
  class WorkerThread extends Thread{
    public void run() {
      //循環(huán)取任務并執(zhí)行
      while(true){ ①
        Runnable task = workQueue.take();
        task.run();
      } 
    }
  }  
}

/** 下面是使用示例 **/
// 創(chuàng)建有界阻塞隊列
BlockingQueue<Runnable> workQueue = 
  new LinkedBlockingQueue<>(2);
// 創(chuàng)建線程池  
MyThreadPool pool = new MyThreadPool(
  10, workQueue);
// 提交任務  
pool.execute(()->{
    System.out.println("hello");
});

常用線程池

ThreadPoolExecutor

在工程中，我們會使用Executors來快速new一個線程池，例如：

ExecutorService executorService = Executors.newFixedThreadPool(threadPoolNum, r -> new Thread(r, threadName));

Executors底層使用的是 ThreadPoolExecutor，我們可以通過ThreadPoolExecutor構造函數來了解ThreadPoolExecutor的一些行為。

ThreadPoolExecutor(
int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)

構造函數參數說明

corePoolSize：表示線程池保有的最小線程數。

maximumPoolSize：表示線程池創(chuàng)建的最大線程數。

keepAliveTime & unit：如果一個線程空閑了keepAliveTime & unit這么久，而且線程池的線程數大于 corePoolSize ，那么這個空閑的線程就要被回收了。

workQueue：工作隊列，和上面示例代碼的工作隊列同義。

threadFactory：通過這個參數你可以自定義如何創(chuàng)建線程，例如你可以給線程指定一個有意義的名字。

handler：通過這個參數你可以自定義任務的拒絕策略。如果線程池中所有的線程都在忙碌，并且工作隊列也滿了（前提是工作隊列是有界隊列），那么此時提交任務，線程池就會拒絕接收。至于拒絕的策略，你可以通過 handler 這個參數來指定。

ThreadPoolExecutor 已經提供了以下 4 種策略。

CallerRunsPolicy：提交任務的線程自己去執(zhí)行該任務。
AbortPolicy：默認的拒絕策略，會 throws RejectedExecutionException。
DiscardPolicy：直接丟棄任務，沒有任何異常拋出。
DiscardOldestPolicy：丟棄最老的任務，其實就是把最早進入工作隊列的任務丟棄，然后把新任務加入到工作隊列。

線程池默認工作行為

不會初始化 corePoolSize 個線程，有任務來了才創(chuàng)建工作線程；

當核心線程滿了之后不會立即擴容線程池，而是把任務堆積到工作隊列中；

當工作隊列滿了后擴容線程池，一直到線程個數達到 maximumPoolSize 為止；（如果線程池還沒有擴容到最大線程數但是工作隊列已經溢出，溢出的請求會被拒絕）

如果隊列已滿且達到了最大線程后還有任務進來，按照拒絕策略處理；

當線程數大于核心線程數時，線程等待 keepAliveTime 后還是沒有任務需要處理的話，收縮線程到核心線程數。

ForkJoinPool

Fork/Join 是一個并行計算的框架，主要就是用來支持分治任務模型的，這個計算框架里的 Fork 對應的是分治任務模型里的任務分解，Join 對應的是結果合并。

Fork/Join 計算框架主要包含兩部分，一部分是分治任務的線程池 ForkJoinPool，另一部分是分治任務 ForkJoinTask。這兩部分的關系類似于 ThreadPoolExecutor 和 Runnable 的關系，都可以理解為提交任務到線程池，只不過分治任務有自己獨特類型 ForkJoinTask。

ForkJoinPool 主要適用于計算密集型任務,Java中的parallelStream底層使用的就是ForkJoinPool。

下面是使用ForkJoinPool的一個簡單例子：

  public static void main(String[] args) {
        ForkJoinPool forkJoinPool = new ForkJoinPool(4);

        Fibonacci fibonacci = new Fibonacci(5);
        Integer res = forkJoinPool.invoke(fibonacci);

        System.out.println(res);
    }

    static class Fibonacci extends RecursiveTask<Integer>{
        final int n;
        Fibonacci(int n){
            this.n = n;
        }
        @Override
        protected Integer compute() {
            if(n<=1){
                return n;
            }
            Fibonacci f1 = new Fibonacci(n-1);
            f1.fork();
            Fibonacci f2 = new Fibonacci(n-2);
            return f2.compute() + f1.join();
        }
    }

FutureTask

我們可以通過FutureTask（Future接口的實現類）獲取線程執(zhí)行結果。FutureTask主要方法如下：

// 取消任務
boolean cancel(
  boolean mayInterruptIfRunning);
// 判斷任務是否已取消  
boolean isCancelled();
// 判斷任務是否已結束
boolean isDone();
// 獲得任務執(zhí)行結果
get();
// 獲得任務執(zhí)行結果，支持超時
get(long timeout, TimeUnit unit);

其中，兩個 get() 方法都是阻塞式的，如果被調用的時候，任務還沒有執(zhí)行完，那么調用 get() 方法的線程會阻塞，直到任務執(zhí)行完才會被喚醒。

ExecutorService executorService = Executors.newFixedThreadPool(10);
        Future<Integer> future = executorService.submit(() -> {
            return 1 + 1;
        });
        Integer res = future.get();
        System.out.println(res);
        Integer res2 = future.get(1000, TimeUnit.SECONDS);
        System.out.println(res2);

FutureTask 實現了 Runnable 和 Future 接口，由于實現了 Runnable 接口，所以可以將 FutureTask 對象作為任務提交給 ThreadPoolExecutor 去執(zhí)行。

// 創(chuàng)建FutureTask
FutureTask<Integer> futureTask
  = new FutureTask<>(()-> 1+2);
// 創(chuàng)建線程池
ExecutorService es = 
  Executors.newCachedThreadPool();
// 提交FutureTask 
es.submit(futureTask);
// 獲取計算結果
Integer result = futureTask.get();

線程數量分析

多線程可以提高程序的響應速度和吞吐量，創(chuàng)建線程的數量會對實際效果產生非常大的影響，線程太少會浪費CPU的資源，線程太多則會導致線程的頻繁切換，系統(tǒng)性能反而會下降。

根據程序類型的不同，我們可以將我們的程序分為IO密集型和CPU密集型兩種，這兩種程序計算最佳線程數的方法有所不同。

CPU密集型

對于 CPU 密集型計算，多線程本質上是提升多核 CPU 的利用率，所以對于一個 4 核的 CPU，每個核一個線程，理論上創(chuàng)建 4 個線程就可以了，再多創(chuàng)建線程也只是增加線程切換的成本。所以，對于 CPU 密集型的計算場景，理論上“線程的數量 =CPU 核數”就是最合適的。不過在工程上，線程的數量一般會設置為“CPU 核數 +1”，這樣的話，當線程因為偶爾的內存頁失效或其他原因導致阻塞時，這個額外的線程可以頂上，從而保證 CPU 的利用率。