[TOC] # ThreadPoolExecutor 無論創建那種線程池 必須要調用ThreadPoolExecutor 線程池類為 java.util.concurrent.ThreadPoolExecutor，常用構造方法為： ~~~ ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) ~~~ ~~~ corePoolSize： 線程池維護線程的最少數量,核心線程池的大小，在線程池被創建之后，其實里面是沒有線程的。（當然，調用prestartAllCoreThreads()或者prestartCoreThread()方法會預創建線程，而不用等著任務的到來）。當有任務進來的時候，才會創建線程。當線程池中的線程數量達到corePoolSize之后，就把任務放到 緩存隊列當中。（就是 workQueue） maximumPoolSize：線程池維護線程的最大數量 ,它標志著這個線程池的最大線程數量。如果沒有最大數量，當創建的線程數量達到了 某個極限值，到最后內存肯定就爆掉了 keepAliveTime： 線程池維護線程所允許的空閑時間,超過這個時間就被終止了。默認情況下，只有 線程池中線程數量 大于 corePoolSize時，keepAliveTime值才會起作用。也就說說，只有在線程池線程數量超出corePoolSize了。我們才會把超時的空閑線程給停止掉。否則就保持線程池中有 corePoolSize 個線程就可以了。 unit： 線程池維護線程所允許的空閑時間的單位,參數keepAliveTime的時間單位，就是 TimeUnit類當中的幾個屬性 workQueue： 線程池所使用的緩沖隊列,用來存儲待執行任務的隊列，不同的線程池它的隊列實現方式不同（因為這關系到排隊策略的問題）比如有以下幾種 ArrayBlockingQueue：基于數組的隊列，創建時需要指定大小。 LinkedBlockingQueue：基于鏈表的隊列，如果沒有指定大小，則默認值是 Integer.MAX_VALUE。（newFixedThreadPool和newSingleThreadExecutor使用的就是這種隊列）。 SynchronousQueue：這種隊列比較特殊，因為不排隊就直接創建新線程把任務提交了。（newCachedThreadPool使用的就是這種隊列）。 threadFactory:線程工廠，用來創建線程 handler： 線程池對拒絕任務的處理策略,絕執行任務時的策略，一般來講有以下四種策略， （1） ThreadPoolExecutor.AbortPolicy 丟棄任務，并拋出 RejectedExecutionException 異常。 （2） ThreadPoolExecutor.CallerRunsPolicy：該任務被線程池拒絕，由調用 execute方法的線程執行該任務。 （3） ThreadPoolExecutor.DiscardOldestPolicy ： 拋棄隊列最前面的任務，然后重新嘗試執行任務。 （4） ThreadPoolExecutor.DiscardPolicy，丟棄任務，不過也不拋出異常。 ~~~ 一個任務通過 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() 拋棄當前的任務 ~~~ 當然也可以根據應用場景實現RejectedExecutionHandler接口，自定義飽和策略，如記錄日志或持久化存儲不能處理的任務。 # 方法 ~~~ shutdown(),平滑的關閉線程池.(如果還有未執行完的任務，就等待它們執行完) shutdownNow(),簡單粗暴的關閉線程池.(沒有執行完的任務也直接關閉) setCorePoolSize(),設置/更改核心池的大小. setMaximumPoolSize(),設置/更改線程池中最大線程的數量限制. getActiveCount() 活躍的線程數 getCorePoolSize() 核心的線程數 getPoolSize() 線程池的大小 getQueue().size() 隊列大小 ~~~ # execute 和 submit execute()：Executor中聲明的方法，向線程池提交一個任務，交由線程池去執行，沒有返回值 submit() ：ExecutorService中聲明的方法。向線程池提交一個任務，交由線程池去執行，可以接受 （ Callable.call() ）回調函數的返回值，適用于需要處理返回著或者異常的業務場景，實際上還是調用的execute()方法，只不過它利用了 Future 來獲取任務執行結果 # schedule schedule: 延時執行任務 scheduleAtFixedRate : 以固定頻率來執行一個任務，按照上一次任務的發起時間計算下一次任務的開始時間 scheduleWithFixedDelay : 以上一次任務的結束時間計算下一次任務的開始時間，在你不能預測調度任務的執行時長時是很有用 # 線程池的關閉 shutdown() ：不會立即終止線程池，而是要等所有任務緩存隊列中的任務都執行完后才終止，但再也不會接受新的任務 shutdownNow() ：立即終止線程池，并嘗試打斷正在執行的任務，并且清空任務緩存隊列，返回尚未執行的任務 # 任務隊列 BlockingQueue BlockingQueue workQueue 取值 ArrayBlockingQueue ：有界的數組隊列，先進先出，此隊列創建時必須指定大小 SynchronousQueue : 同步的阻塞隊列，它不會保存提交的任務，而是將直接新建一個線程來執行新來的任務 LinkedBlockingQueue ：基于鏈表的先進先出隊列，可支持有界/無界的隊列，如果創建時沒有指定此隊列大小，則默認為Integer.MAX_VALUE PriorityBlockingQueue ：優先隊列，可以針對任務排序 # 自定義線程池 ~~~ package com.study; import java.util.Locale; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; public class ExecutorServiceHelper { /** * 獲取活躍的cpu數量 */ private static int NUMBER_OF_CORES = Runtime.getRuntime().availableProcessors(); private final static BlockingQueue<Runnable> queue; private final static long KEEP_ALIVE_TIME = 3L; private final static TimeUnit KEEP_ALIVE_TIME_UNIT = TimeUnit.SECONDS; private static ThreadFactory mThreadFactory; static { /** * 基于鏈表的隊列，如果沒有指定大小，則默認值是 Integer.MAX_VALUE */ queue = new LinkedBlockingQueue<Runnable>(); //默認的工廠方法將新創建的線程命名為：pool-[虛擬機中線程池編號]-thread-[線程編號] //mThreadFactory= Executors.defaultThreadFactory(); mThreadFactory = new NamedThreadFactory(); //System.out.println("NUMBER_OF_CORES:"+NUMBER_OF_CORES); } public static void execute(Runnable runnable) { if (runnable == null) { return; } /** * 1.當線程池小于 corePoolSize 時，新提交任務將創建一個新線程執行任務，即使此時線程池中存在空閑線程。 * 2.當線程池達到 corePoolSize 時，新提交任務將被放入 workQueue 中，等待線程池中任務調度執行 * 3.當 workQueue 已滿，且 maximumPoolSize > corePoolSize時，新提交任務會創建新線程執行任務 * 4.當提交任務數超過 maximumPoolSize 時，新提交任務由 RejectedExecutionHandler 處理 * 5.當線程池中超過 corePoolSize 線程，空閑時間達到 keepAliveTime 時，關閉空閑線程 * 6.當設置 allowCoreThreadTimeOut(true) 時，線程池中 corePoolSize 線程空閑時間達到 keepAliveTime 也將關閉 **/ /** maximumPoolSize 推薦取值 如果是 CPU 密集型任務，就需要盡量壓榨CPU，參考值可以設為 NUMBER_OF_CORES + 1 或 NUMBER_OF_CORES + 2 如果是 IO 密集型任務，參考值可以設置為 NUMBER_OF_CORES * 2 */ ExecutorService executorService = new ThreadPoolExecutor(NUMBER_OF_CORES, NUMBER_OF_CORES * 2, KEEP_ALIVE_TIME, KEEP_ALIVE_TIME_UNIT, queue, mThreadFactory); executorService.execute(runnable); } private static class NamedThreadFactory implements ThreadFactory { private final AtomicInteger threadNumberAtomicInteger = new AtomicInteger(1); @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r, String.format(Locale.CHINA, "%s%d", "NamedThreadFactory", threadNumberAtomicInteger.getAndIncrement())); /* thread.setDaemon(true);//是否是守護線程 thread.setPriority(Thread.NORM_PRIORITY);//設置優先級 1~10 有3個常量 默認 Thread.MIN_PRIORITY*/ return thread; } } } ~~~ 測試下 ~~~ class TestThread { public static void main(String[] args) { ExecutorServiceHelper.execute(new Runnable() { @Override public void run() { //do something System.out.println("1"); } }); } } ~~~