Java並發編程:線程池ThreadPoolExecutor


  多線程的程序的確能發揮多核處理器的性能。雖然與進程相比,線程輕量化了很多,但是其創建和關閉同樣需要花費時間。而且線程多了以后,也會搶占內存資源。如果不對線程加以管理的話,是一個非常大的隱患。而線程池的目的就是管理線程。當你需要一個線程時,你就可以拿一個空閑線程去執行任務,當任務執行完后,線程又會歸還到線程池。這樣就有效的避免了重復創建、關閉線程和線程數量過多帶來的問題。

Java並發包提供的線程池

 

注:摘自《實戰Java高並發程序設計》

  如圖是Java並發包下提供的線程池功能。其中ExecutorService接口提供一些操作線程池的方法。而Executors相當於一個線程池工廠類,它里面有幾種現成的具備某種特定功能的線程池工廠方法。看到這些應該不陌生,舉個我們平時最常使用的例子:

//創建一個大小為10的固定線程池
ExecutorService threadpool= Executors.newScheduledThreadPool(10);

  下面簡單介紹一下這些工廠方法:

  newFixedThreadPool()方法:固定線程數量線程池。傳入的數字就是線程的數量,如果有空閑線程就去執行任務,如果沒有空閑線程就會把任務放到一個任務隊列,等到有線程空閑時便去處理隊列中的任務。

  newSingleThreadExecutor()方法:只有一個線程的線程池。同樣,超出的任務會被放到任務隊列,等這個線程空閑時就會去按順序處理。

  newCachedThreadPool()方法:可以根據實際情況拓展的線程池。當沒有空閑線程去執行新任務時,就會再創建新的線程去執行任務,執行完后新建的線程也會返回線程池進行復用。

  newSingleThreadScheduledExecutor()方法:返回的是ScheduledExecutorService對象。ScheduledExecutorService是繼承於ExecutorService的,有一些拓展方法,如指定執行時間。這個線程池大小為1,在指定時間執行任務。關於指定時間的幾個方法:schedule()是在指定時間后執行一次任務。scheduleAtFixedRate()和方法scheduleWithFixedDelay()方法,兩者都是周期性的執行任務,但是前者是以上一次任務開始為周期起點,后者是以上一次任務結束為周期起點。具體的參數大家可以在IDE里面查看。

  newScheduledThreadPool()方法:和上面一個方法一樣,但是可以指定線程池大小,其實上面那個方法也是調用這個方法的,只是傳入的參數是1。

線程池核心類

  上面簡單的對Java並發包下線程池的結構和API進行簡單的介紹,下面開始深入了解一下線程池。如果大家在IDE上追蹤一下上面幾個工廠方法就會發現,其中最后都會調用一個方法,通過上圖其實也可以發現。那就是ThreadPoolExecutor的構造方法,工廠方法只是幫我們傳入不同的參數,從而實現不同的效果,所以如果你想更自由的控制自己的線程池,推薦直接使用ThreadPoolExecutor創建線程池。下面給出這個構造函數的參數列表:

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

  參數從上到下,作用依次為:

  1.指定線程池種線程的數量。

  2.線程池種最大的線程數量,也就是最大能拓展到多少。

  3.當線程數量超過corePoolSize,多余的空閑線程多久會被銷毀。

  4.keepAliveTime的單位。

  5.任務隊列,當空閑線程不夠,也不能再新建線程時,新提交的任務就會被放到任務隊列種。

  6.線程工廠,用於創建線程,默認的即可。

  7.拒絕策略。當任務太多,達到最大線程數量、任務隊列也滿了,該如何拒絕新提交的任務。

任務隊列

  任務隊列是一個BlockingQueue接口,在ThreadPoolExecutor一共有如下幾種實現類實現了BlockingQueue接口。

  SynchronousQueue:直接提交隊列。這種隊列其實不會真正的去保存任務,每提交一個任務就直接讓空閑線程執行,如果沒有空閑線程就去新建,當達到最大線程數時,就會執行拒絕策略。所以使用這種任務隊列時,一般會設置很大的maximumPoolSize,不然很容易就執行了拒絕策略。newCachedThreadPool線程池的corePoolSize為0,maximumPoolSize無限大,它用的就是直接提交隊列。

  ArrayBlockingQueue:有界任務隊列,其構造函數必須帶一個容量參數,表示任務隊列的大小。當線程數量小於corePoolSize時,有任務進來優先創建線程。當線程數等於corePoolSize時,新任務就會進入任務隊列,當任務隊列滿了,才會創建新線程,線程數達到maximumPoolSize時執行拒絕策略。

  LinkedBlockingQueue:無界任務隊列,通過它的名字也應該知道了,它是個鏈表,除非沒有空間了,不然不會出現任務隊列滿了的情況,但是非常耗費系統資源。和有界任務隊列一樣,線程數若小於corePoolSize,新任務進來時沒有空閑線程的話就會創建新線程,當達到corePoolSize時,就會進入任務隊列。會發現沒有maximumPoolSize什么事,newFixedThreadPool固定大小線程池就是用的這個任務隊列,它的corePoolSize和maximumPoolSize相等。

  PriorityBlockingQueue:優先任務隊列,它是一個特殊的無界隊列,因為它總能保證高優先級的任務先執行。

拒絕策略

  JDK提供了四種拒絕策略。

  AbortPolicy:直接拋出異常,阻止系統正常工作。

  CallerRunsPolicy:如果線程池未關閉,則在調用者線程里面執行被丟棄的任務,這個策略不是真正的拒絕任務。比如我們在T1線程中提交的任務,那么該拒絕策略就會把多余的任務放到T1線程執行,會影響到提交者線程的性能。

  DiscardOldestPolicy:該策略會丟棄一個最老的任務,也就是即將被執行的任務,然后再次嘗試提交該任務。

  DiscardPolicy:直接丟棄多余的任務,不做任何處理,如果允許丟棄任務,這個策略是最好的。

  以上內置的拒絕策略都實現了RejectedExecutionHandler接口,所以上面的拒絕策略無法滿足你的要求,可以自定義一個:繼承RejectedExecutionHandler並實現rejectedExecution方法。

線程工廠

線程池中的線程是由ThreadFactory負責創建的,一般情況下默認就行,如果有一些其他的需求,比如自定義線程的名稱、優先級等,我們也可以利用ThreadFactory接口來自定義自己的線程工廠:繼承ThreadFactory並實現newThread方法。

線程池的拓展

  在ThreadPoolExecutor中有三個擴展方法:分別會在任務執行前beforeExecute、執行完成afterExecute、線程池退出時執行terminated。

  這幾個方法在哪調用的?在ThreadPoolExecutor中有一個內部類:Worker,每個線程的任務其實都是由這個類里面的run方法執行的,貼一下這個類的源碼:

 

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;

    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;
    //....省略
    
    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }
    //....省略
}

 

  接着進入這個runWorker方法:

final void runWorker(Worker w) {
    //...省略
            try {
                //任務執行前
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    //任務執行完
                    afterExecute(task, thrown);
                }
            } 
    //....省略
}

  還有一個線程池退出時執行的方法是在何處執行的?這個方法被調用的地方就不止一處了,像線程池的shutdown方法就會調用

public void shutdown() {
    //....省略。這個方法里面就會調用terminated
    tryTerminate();
}

  ThreadPoolExecutor中這三個方法默認是沒有任何內容的,所以我們要自定義它也很簡單,直接重寫它們就行了:

ExecutorService threadpool= new ThreadPoolExecutor(5,5,0L,TimeUnit.SECONDS,new LinkedBlockingDeque<>()){
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        //執行任務前
    }
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        //執行任務后
    }
    @Override
    protected void terminated() {
        //線程退出
    }
};

 

打賞

免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2021 CODEPRJ.COM