快捷導(dǎo)航

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

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

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

線程池模型

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

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

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

//簡化的線程池，僅用來說明工作原理
class MyThreadPool{
  //利用阻塞隊列實現(xiàn)生產(chǎn)者-消費者模式
  BlockingQueue<Runnable> workQueue;
  //保存內(nèi)部工作線程
  List<WorkerThread> threads 
    = new ArrayList<>();
  // 構(gòu)造方法
  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);
    }
  }
  // 提交任務(wù)
  void execute(Runnable command){
    workQueue.put(command);
  }
  // 工作線程負(fù)責(zé)消費任務(wù)，并執(zhí)行任務(wù)
  class WorkerThread extends Thread{
    public void run() {
      //循環(huán)取任務(wù)并執(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);
// 提交任務(wù)  
pool.execute(()->{
    System.out.println("hello");
});

常用線程池

ThreadPoolExecutor

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

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

Executors底層使用的是 ThreadPoolExecutor，我們可以通過ThreadPoolExecutor構(gòu)造函數(shù)來了解ThreadPoolExecutor的一些行為。

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

構(gòu)造函數(shù)參數(shù)說明

corePoolSize：表示線程池保有的最小線程數(shù)。

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

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

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

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

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

ThreadPoolExecutor 已經(jīng)提供了以下 4 種策略。

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

線程池默認(rèn)工作行為

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

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

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

如果隊列已滿且達(dá)到了最大線程后還有任務(wù)進(jìn)來，按照拒絕策略處理；

當(dāng)線程數(shù)大于核心線程數(shù)時，線程等待 keepAliveTime 后還是沒有任務(wù)需要處理的話，收縮線程到核心線程數(shù)。

ForkJoinPool

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

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

ForkJoinPool 主要適用于計算密集型任務(wù),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接口的實現(xiàn)類）獲取線程執(zhí)行結(jié)果。FutureTask主要方法如下：

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

其中，兩個 get() 方法都是阻塞式的，如果被調(diào)用的時候，任務(wù)還沒有執(zhí)行完，那么調(diào)用 get() 方法的線程會阻塞，直到任務(wù)執(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 實現(xiàn)了 Runnable 和 Future 接口，由于實現(xiàn)了 Runnable 接口，所以可以將 FutureTask 對象作為任務(wù)提交給 ThreadPoolExecutor 去執(zhí)行。

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

線程數(shù)量分析

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

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

CPU密集型

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