Java 併發--- 線程池

時間  2019-12-08
標籤 java 併發 線程 欄目 Java 简体版
原文   原文鏈接

ThreadPoolExecutor

ThreadPoolExecutor繼承了AbstractExecutorService類，並提供了四個構造器，事實上，經過觀察每一個構造器的源碼具體實現，發現前面三個構造器都是調用的第四個構造器進行的初始化工做。java

public class ThreadPoolExecutor extends AbstractExecutorService {
    .....
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
        BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);
    .....
}

各個參數含義：數組

corePoolSize：核心池的大小。在建立了線程池後，默認狀況下，線程池中並無任何線程，而是等待有任務到來才建立線程去執行任務，除非調用了prestartAllCoreThreads()或者prestartCoreThread()方法，從這2個方法的名字就能夠看出，是預建立線程的意思，即在沒有任務到來以前就建立corePoolSize個線程或者一個線程。默認狀況下，在建立了線程池後，線程池中的線程數爲0，當有任務來以後，就會建立一個線程去執行任務，當線程池中的線程數目達到corePoolSize後，就會把到達的任務放到緩存隊列當中。緩存

maximumPoolSize：線程池最大線程數。它表示在線程池中最多能建立多少個線程。ide

keepAliveTime：表示線程沒有任務執行時最多保持多久時間會終止。默認狀況下，只有當線程池中的線程數大於corePoolSize時，keepAliveTime纔會起做用，直到線程池中的線程數不大於corePoolSize。即當線程池中的線程數大於corePoolSize時，若是一個線程空閒的時間達到keepAliveTime，則會終止，直到線程池中的線程數不超過corePoolSize。可是若是調用了allowCoreThreadTimeOut(boolean)方法，在線程池中的線程數不大於corePoolSize時，keepAliveTime參數也會起做用，直到線程池中的線程數爲0；this

unit：參數keepAliveTime的時間單位，有7種取值，在TimeUnit類中有7種靜態屬性。線程

workQueue：一個阻塞隊列，用來存儲等待執行的任務，這個參數的選擇也很重要，會對線程池的運行過程產生重大影響，通常來講，這裏的阻塞隊列有如下幾種選擇：ArrayBlockingQueue;
LinkedBlockingQueue; SynchronousQueue; PriorityBlockingQueuerest

threadFactory：線程工廠，主要用來建立線程；code

handler：表示當拒絕處理任務時的策略。繼承

ThreadPoolExecutor繼承了AbstractExecutorService，AbstractExecutorService是一個抽象類，它實現了ExecutorService接口，而ExecutorService又是繼承了Executor接口。Executor接口定義：接口

public interface Executor {
    void execute(Runnable command);
}

Executor是一個頂層接口，在它裏面只聲明瞭一個方法execute(Runnable command)，返回值爲void，參數爲Runnable類型，從字面意思能夠理解，就是用來執行傳進去的任務的；

而後ExecutorService接口繼承了Executor接口，並聲明瞭一些方法：submit、invokeAll、invokeAny以及shutDown等；

抽象類AbstractExecutorService實現了ExecutorService接口，基本實現了ExecutorService中聲明的全部方法；

最後ThreadPoolExecutor繼承了類AbstractExecutorService。

在ThreadPoolExecutor類中有幾個很是重要的方法：

execute()
submit()
shutdown()
shutdownNow()

execute()方法其實是Executor中聲明的方法，在ThreadPoolExecutor進行了具體的實現，這個方法是ThreadPoolExecutor的核心方法，經過這個方法能夠向線程池提交一個任務，交由線程池去執行。

submit()方法是在ExecutorService中聲明的方法，在AbstractExecutorService就已經有了具體的實現，在ThreadPoolExecutor中並無對其進行重寫，這個方法也是用來向線程池提交任務的，可是它和execute()方法不一樣，它可以返回任務執行的結果，去看submit()方法的實現，會發現它實際上仍是調用的execute()方法，只不過它利用了Future來獲取任務執行結果。

shutdown()和shutdownNow()是用來關閉線程池的。

其餘方法：getQueue() 、getPoolSize() 、getActiveCount()、getCompletedTaskCount()等獲取與線程池相關屬性的方法。

線程池實現原理

線程池狀態

在ThreadPoolExecutor中定義了一個volatile變量，另外定義了幾個static final變量表示線程池的各個狀態：

volatile int runState;
static final int RUNNING    = 0;
static final int SHUTDOWN   = 1;
static final int STOP       = 2;
static final int TERMINATED = 3;

runState表示當前線程池的狀態，它是一個volatile變量用來保證線程之間的可見性；下面的幾個static final變量表示runState可能的幾個取值。

當建立線程池後，初始時，線程池處於RUNNING狀態；

若是調用了shutdown()方法，則線程池處於SHUTDOWN狀態，此時線程池不可以接受新的任務，它會等待全部任務執行完畢；

若是調用了shutdownNow()方法，則線程池處於STOP狀態，此時線程池不能接受新的任務，而且會去嘗試終止正在執行的任務；

當線程池處於SHUTDOWN或STOP狀態，而且全部工做線程已經銷燬，任務緩存隊列已經清空或執行結束後，線程池被設置爲TERMINATED狀態。

任務的執行

ThreadPoolExecutor類中其餘的一些比較重要成員變量:

private final BlockingQueue<Runnable> workQueue;              //任務緩存隊列，用來存放等待執行的任務
private final ReentrantLock mainLock = new ReentrantLock();   //線程池的主要狀態鎖，對線程池狀態（好比線程池大小、runState等）的改變都要使用這個鎖
private final HashSet<Worker> workers = new HashSet<Worker>();  //用來存放工做集
 
private volatile long  keepAliveTime;    //線程存貨時間   
private volatile boolean allowCoreThreadTimeOut;   //是否容許爲核心線程設置存活時間
private volatile int   corePoolSize;     //核心池的大小（即線程池中的線程數目大於這個參數時，提交的任務會被放進任務緩存隊列）
private volatile int   maximumPoolSize;   //線程池最大能容忍的線程數
 
private volatile int   poolSize;       //線程池中當前的線程數
 
private volatile RejectedExecutionHandler handler; //任務拒絕策略
 
private volatile ThreadFactory threadFactory;   //線程工廠，用來建立線程
 
private int largestPoolSize;   //用來記錄線程池中曾經出現過的最大線程數
 
private long completedTaskCount;   //用來記錄已經執行完畢的任務個數

舉例：

假若有一個工廠，工廠裏面有10個工人，每一個工人同時只能作一件任務。所以只要當10個工人中有工人是空閒的，來了任務就分配給空閒的工人作；

當10個工人都有任務在作時，若是還來了任務，就把任務進行排隊等待；

若是說新任務數目增加的速度遠遠大於工人作任務的速度，那麼此時工廠主管可能會想補救措施，好比從新招4個臨時工人進來；而後就將任務也分配給這4個臨時工人作；

若是說着14個工人作任務的速度仍是不夠，此時工廠主管可能就要考慮再也不接收新的任務或者拋棄前面的一些任務了。

當這14個工人當中有人空閒時，而新任務增加的速度又比較緩慢，工廠主管可能就考慮辭掉4個臨時工了，只保持原來的10個工人，畢竟請額外的工人是要花錢的。

這個例子中的corePoolSize就是10，而maximumPoolSize就是14（10+4）。

也就是說corePoolSize就是線程池大小，maximumPoolSize是線程池的一種補救措施，即任務量忽然過大時的一種補救措施。

largestPoolSize只是一個用來起記錄做用的變量，用來記錄線程池中曾經有過的最大線程數目，跟線程池的容量沒有任何關係。

在ThreadPoolExecutor類中，最核心的任務提交方法是execute()方法，雖然經過submit也能夠提交任務，可是實際上submit方法裏面最終調用的仍是execute()方法，因此只須要研究execute()方法的實現原理便可。

// TODO

線程池中的線程初始化

默認狀況下，建立線程池以後，線程池中是沒有線程的，須要提交任務以後纔會建立線程。在實際中若是須要線程池建立以後當即建立線程，能夠經過如下兩個方法辦到：

prestartCoreThread()：初始化一個核心線程；
prestartAllCoreThreads()：初始化全部核心線程

任務緩存隊列及排隊策略

任務緩存隊列，即workQueue，它用來存放等待執行的任務。

workQueue的類型爲BlockingQueue ，一般能夠取下面三種類型：

  • ArrayBlockingQueue：基於數組的先進先出隊列，此隊列建立時必須指定大小；
  • LinkedBlockingQueue：基於鏈表的先進先出隊列，若是建立時沒有指定此隊列大小，則默認爲Integer.MAX_VALUE；
  • synchronousQueue：這個隊列比較特殊，它不會保存提交的任務，而是將直接新建一個線程來執行新來的任務。

任務拒絕策略

當線程池的任務緩存隊列已滿而且線程池中的線程數目達到maximumPoolSize，若是還有任務到來就會採起任務拒絕策略，一般有如下四種策略：

ThreadPoolExecutor.AbortPolicy:丟棄任務並拋出RejectedExecutionException異常。
ThreadPoolExecutor.DiscardPolicy：也是丟棄任務，可是不拋出異常。
ThreadPoolExecutor.DiscardOldestPolicy：丟棄隊列最前面的任務，而後從新嘗試執行任務（重複此過程）
ThreadPoolExecutor.CallerRunsPolicy：由調用線程處理該任務

線程池的關閉

ThreadPoolExecutor提供了兩個方法，用於線程池的關閉，分別是shutdown()和shutdownNow()，其中：

  • shutdown()：不會當即終止線程池，而是要等全部任務緩存隊列中的任務都執行完後才終止，但不再會接受新的任務
  • shutdownNow()：當即終止線程池，並嘗試打斷正在執行的任務，而且清空任務緩存隊列，返回還沒有執行的任務

線程池容量的動態調整

ThreadPoolExecutor提供了動態調整線程池容量大小的方法：setCorePoolSize()和setMaximumPoolSize().

使用示例

public class Test {
     public static void main(String[] args) {   
         ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
                 new ArrayBlockingQueue<Runnable>(5));
          
         for(int i=0;i<15;i++){
             MyTask myTask = new MyTask(i);
             executor.execute(myTask);
             System.out.println("線程池中線程數目："+executor.getPoolSize()+"，隊列中等待執行的任務數目："+
             executor.getQueue().size()+"，已執行玩別的任務數目："+executor.getCompletedTaskCount());
         }
         executor.shutdown();
     }
}
 
class MyTask implements Runnable {
    private int taskNum;
     
    public MyTask(int num) {
        this.taskNum = num;
    }
     
    @Override
    public void run() {
        System.out.println("正在執行task "+taskNum);
        try {
            Thread.currentThread().sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("task "+taskNum+"執行完畢");
    }
}

執行結果：

正在執行task 0
線程池中線程數目：1，隊列中等待執行的任務數目：0，已執行玩別的任務數目：0
線程池中線程數目：2，隊列中等待執行的任務數目：0，已執行玩別的任務數目：0
正在執行task 1
線程池中線程數目：3，隊列中等待執行的任務數目：0，已執行玩別的任務數目：0
正在執行task 2
線程池中線程數目：4，隊列中等待執行的任務數目：0，已執行玩別的任務數目：0
正在執行task 3
線程池中線程數目：5，隊列中等待執行的任務數目：0，已執行玩別的任務數目：0
正在執行task 4
線程池中線程數目：5，隊列中等待執行的任務數目：1，已執行玩別的任務數目：0
線程池中線程數目：5，隊列中等待執行的任務數目：2，已執行玩別的任務數目：0
線程池中線程數目：5，隊列中等待執行的任務數目：3，已執行玩別的任務數目：0
線程池中線程數目：5，隊列中等待執行的任務數目：4，已執行玩別的任務數目：0
線程池中線程數目：5，隊列中等待執行的任務數目：5，已執行玩別的任務數目：0
線程池中線程數目：6，隊列中等待執行的任務數目：5，已執行玩別的任務數目：0
正在執行task 10
線程池中線程數目：7，隊列中等待執行的任務數目：5，已執行玩別的任務數目：0
正在執行task 11
線程池中線程數目：8，隊列中等待執行的任務數目：5，已執行玩別的任務數目：0
正在執行task 12
線程池中線程數目：9，隊列中等待執行的任務數目：5，已執行玩別的任務數目：0
正在執行task 13
線程池中線程數目：10，隊列中等待執行的任務數目：5，已執行玩別的任務數目：0
正在執行task 14
task 3執行完畢
task 0執行完畢
task 2執行完畢
task 1執行完畢
正在執行task 8
正在執行task 7
正在執行task 6
正在執行task 5
task 4執行完畢
task 10執行完畢
task 11執行完畢
task 13執行完畢
task 12執行完畢
正在執行task 9
task 14執行完畢
task 8執行完畢
task 5執行完畢
task 7執行完畢
task 6執行完畢
task 9執行完畢

從執行結果能夠看出，當線程池中線程的數目大於5時，便將任務放入任務緩存隊列裏面，當任務緩存隊列滿了以後，便建立新的線程。若是上面程序中，將for循環中改爲執行20個任務，就會拋出任務拒絕異常了。

不過在java doc中，並不提倡咱們直接使用ThreadPoolExecutor，而是使用Executors類中提供的幾個靜態方法來建立線程池：

Executors.newCachedThreadPool();       //建立一個緩衝池，緩衝池容量大小爲Integer.MAX_VALUE
Executors.newSingleThreadExecutor();   //建立容量爲1的緩衝池
Executors.newFixedThreadPool(int);     //建立固定容量大小的緩衝池

下面是這三個靜態方法的具體實現:

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

newFixedThreadPool建立的線程池corePoolSize和maximumPoolSize值是相等的，它使用LinkedBlockingQueue；

newSingleThreadExecutor將corePoolSize和maximumPoolSize都設置爲1，也使用LinkedBlockingQueue；

newCachedThreadPool將corePoolSize設置爲0，將maximumPoolSize設置爲Integer.MAX_VALUE，使用SynchronousQueue，也就是說來了任務就建立線程運行，當線程空閒超過60秒，就銷燬線程。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程