線程池ThreadPoolExcutor的使用

 

摘自阿里巴巴開發手冊：

　　　　　　【強制】線程池不允許使用 Executors 去創建，而是通過 ThreadPoolExecutor 的方式，這樣 的處理方式讓寫的同學更加明確線程池的運行規則，規避資源耗盡的風險。 說明：Executors 返回的線程池對象的弊端如下： 1）FixedThreadPool 和 SingleThreadPool: 允許的請求隊列長度為 Integer.MAX_VALUE，可能會堆積大量的請求，從而導致 OOM。 2）CachedThreadPool 和 ScheduledThreadPool: 允許的創建線程數量為 Integer.MAX_VALUE，可能會創建大量的線程，從而導致 OOM。

 

摘自網絡：

1.優勢: 
(1)降低資源消耗。通過重復利用已創建的線程降低線程創建、銷毀線程造成的消耗。 
(2)提高響應速度。當任務到達時，任務可以不需要等到線程創建就能立即執行。 
(3)提高線程的可管理性。線程是稀缺資源，如果入限制的創建，不僅會消耗系統資源，還會降低系統的穩定性，使用線程池可以進行統一的分配、調優和監控。

2.線程池的創建 
new ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,long keepAliveTime, TimeUnit unit,BlockingQueue workQueue,RejectedExecutionHandler handler) 
(1)corePoolSize： 線程池維護線程的最少數量 （core : 核心） 
(2)maximumPoolSize： 線程池維護線程的最大數量 
(3)keepAliveTime： 線程池維護線程所允許的空閑時間 
(4)unit： 線程池維護線程所允許的空閑時間的單位 
(5)workQueue： 線程池所使用的緩沖隊列 
(6)handler： 線程池對拒絕任務的處理策略

3.添加任務到線程池 
通過 execute(Runnable)方法被添加到線程池，任務就是一個 Runnable類型的對象，任務的執行方法就是 Runnable類型對象的run()方法。 
當一個任務通過execute(Runnable)方法欲添加到線程池時： 
如果此時線程池中的數量小於corePoolSize，即使線程池中的線程都處於空閑狀態，也要創建新的線程來處理被添加的任務。 
如果此時線程池中的數量等於 corePoolSize，但是緩沖隊列 workQueue未滿，那么任務被放入緩沖隊列。 
如果此時線程池中的數量大於corePoolSize，緩沖隊列workQueue滿，並且線程池中的數量小於maximumPoolSize，建新的線程來處理被添加的任務。 
如果此時線程池中的數量大於corePoolSize，緩沖隊列workQueue滿，並且線程池中的數量等於maximumPoolSize，那么通過 handler所指定的策略來處理此任務。 
也就是：處理任務的優先級為： 
核心線程corePoolSize、任務隊列workQueue、最大線程maximumPoolSize，如果三者都滿了，使用handler處理被拒絕的任務。 
當線程池中的線程數量大於 corePoolSize時，如果某線程空閑時間超過keepAliveTime，線程將被終止。這樣，線程池可以動態的調整池中的線程數。 
unit可選的參數為java.util.concurrent.TimeUnit中的幾個靜態屬性：NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS。 
workQueue常用的是：java.util.concurrent.ArrayBlockingQueue 
handler有四個選擇： 
ThreadPoolExecutor.AbortPolicy()： 拋出java.util.concurrent.RejectedExecutionException異常 
ThreadPoolExecutor.CallerRunsPolicy(): 重試添加當前的任務，他會自動重復調用execute()方法 
ThreadPoolExecutor.DiscardOldestPolicy(): 拋棄舊的任務 
ThreadPoolExecutor.DiscardPolicy(): 拋棄當前的任務

4.線程池的使用場合 
（1）單個任務處理的時間比較短； 
（2）需要處理的任務數量大；

6、線程池的應用舉例：

package hh;
import java.io.Serializable;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolExcutorDemo {
     private static int produceTaskSleepTime = 5;
     private static int consumeTaskSleepTime = 5000;
     private static int produceTaskMaxNumber = 20; //定義最大添加10個線程到線程池中
     public static void main(String[] args) {
            //構造一個線程池
           ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3,
                     TimeUnit. SECONDS, new ArrayBlockingQueue<Runnable>(3),
                      new ThreadPoolExecutor.DiscardOldestPolicy());
            for( int i=1; i<= produceTaskMaxNumber;i++){
                 try {
                      //一個任務，並將其加入到線程池
                     String work= "work@ " + i;
                     System. out.println( "put ：" +work);
                      threadPool.execute( new ThreadPoolTask(work));
                      //便於觀察，等待一段時間
                     Thread. sleep(produceTaskSleepTime);
                } catch (Exception e) {
                      e.printStackTrace();
                     }
                }
     }

     /**
     * 線程池執行的任務
     * @author zhu
     */
     public static class ThreadPoolTask implements Runnable,Serializable{
            private static final long serialVersionUID = 0;
            //保存任務所需要的數據
            private Object threadPoolTaskData;
           ThreadPoolTask(Object works){
                 this. threadPoolTaskData =works;
           }
            public void run(){
                 //處理一個任務，這里的處理方式太簡單了，僅僅是一個打印語句
                System. out.println( "start------"+threadPoolTaskData );
                 try {
                      //便於觀察，等待一段時間
                     Thread. sleep(consumeTaskSleepTime);
                     } catch (Exception e) {
                            e.printStackTrace();
                           }
                 threadPoolTaskData = null;
                }
            public Object getTask(){
                 return this. threadPoolTaskData;
                }
     }
}