java內置線程池ThreadPoolExecutor源碼學習記錄
背景
公司業務性能優化，使用java自帶的Executors.newFixedThreadPool()方法生成線程池。但是其內部定義的LinkedBlockingQueue容量是Integer.MAX_VALUE。考慮到如果數據庫中待處理數據量很大有可能會在短時間內往LinkedBlockingQueue中填充很多數據，導致內存溢出。於是看了一下線程池這塊的源碼，並在此記錄。
類圖
-
Executor是一個頂層接口，在它裏面只聲明了一個方法execute(Runnable)，返回值為void，參數為Runnable類型，從字面意思可以理解，就是用來執行傳進去的任務的；
-
ExecutorService接口繼承了Executor接口，並聲明了一些方法：submit、invokeAll、invokeAny以及shutDown等
-
抽象類AbstractExecutorService實現了ExecutorService接口，基本實現了ExecutorService中聲明的所有方法；submit() 方法
-
ThreadPoolExecutor繼承了類AbstractExecutorService。實現了execute(Runnable)方法。
-
Executors提供的集中工廠方法都是調用的ThreadPoolExecutor的構造方法。因為這個構造方法參數比較多 所以提供了幾個經典的實現。
ExecutorService newCachedThreadPool = Executors.newFixedThreadPool(); ExecutorService newCachedThreadPool= Executors.newSingleThreadExecutor(); ExecutorService newCachedThreadPool = Executors.newCachedThreadPool(); ExecutorService newCachedThreadPool = Executors.newScheduledThreadPool();
-
本篇違章主要包括以下幾點內容。這也是解決背景中提到的問題的主要歷程。
1.ThreadPoolExecutor構造方法
2.ExecutorService submit() 方法的實現
2.Executor execute() 方法的實現
3.reject() 拒絕策略
ThreadPoolExecutor構造方法
構造方法中賦值的成員標量：
// 構造方法中用到的成員變量 private volatile int corePoolSize; //核心池的大小（即線程池中的線程數目大於這個參數時，提交的任務會被放進任務緩存隊列） private volatile int maximumPoolSize; //線程池最大能容忍的線程數 private volatile long keepAliveTime; //線程空閑之後存貨時間 （線程數量大於corePoolSize之後） private final BlockingQueue<Runnable> workQueue; //任務緩存隊列，用來存放等待執行的任務 private volatile ThreadFactory threadFactory; //線程工廠，用來創建線程 private volatile RejectedExecutionHandler handler; //任務拒絕策略
通過代碼可以知道 Executors提供的集中工廠方法實際都是調用的同一個ThreadPoolExecutor的構造方法。當然我們也可以通過自己調用ThreadPoolExecutor構造方法 自己設置參數 從而獲得很貼合我們業務的線程池。
AbstractExecutorService submit() 方法
/** * @throws RejectedExecutionException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */ public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask; }
其實是調用了execute() 方法，execute()方法 由ThreadPoolExecutor類實現。
ThreadPoolExecutor execute()方法
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 29位 private static final int COUNT_BITS = Integer.SIZE - 3; // 0001 1111 1111 1111 1111 1111 1111 1111 private static final int CAPACITY = (1 << COUNT_BITS) - 1; // runState is stored in the high-order bits private static final int RUNNING = -1 << COUNT_BITS; private static final int SHUTDOWN = 0 << COUNT_BITS; private static final int STOP = 1 << COUNT_BITS; private static final int TIDYING = 2 << COUNT_BITS; private static final int TERMINATED = 3 << COUNT_BITS; // Packing and unpacking ctl // 高三位 代表 狀態 private static int runStateOf(int c) { return c & ~CAPACITY; } // 低三位 代表 數量 private static int workerCountOf(int c) { return c & CAPACITY; } // 把狀態和數量兩個值 揉在一起 // private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static int ctlOf(int rs, int wc) { return rs | wc; }
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); // 獲取到當前有效的線程數和線程池的狀態 int c = ctl.get(); // 1.獲取當前正在運行線程數是否小於核心線程池，是則新創建一個線程執行任務，否則將任務放到任務隊列中 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // 2.當前核心線程池中全部線程都在運行workerCountOf(c) >= corePoolSize，所以此時將線程放到任務隊列中 // 線程池是否處於運行狀態，且是否任務插入任務隊列成功。註意這塊 && 是做了優化如果前面條件失敗後面語句不會處理 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); //在此檢查線程池是否處於運行狀態，如果不是則使剛剛的任務出隊。和上面一樣 && 是做了優化如果前面條件失敗後面語句不會處理 if (! isRunning(recheck) && remove(command)) reject(command); // 如果沒有執行的線程，就再開啟一個線程（有可能沒有核心線程） else if (workerCountOf(recheck) == 0) addWorker(null, false); } // 3.插入隊列不成功 offer() 方法失敗是因為隊列滿了，此時就新創建線程去執行任務，創建失敗拋出異常 else if (!addWorker(command, false)) reject(command); }
// CAS修改clt的值+1，成功退出cas循環，失敗繼續 if (compareAndIncrementWorkerCount(c)) break retry;
//將新建的線程加入到線程池中 workers.add(w); int s = workers.size(); //修正largestPoolSize的值 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true;
addWorker()方法 總結起來就兩部分
1.CAS+失敗重試操作來將線程數加1
2.新建一個線程並啟用。
RejectedExecutionHandler拒絕策略
java 內置的四種拒絕策略。
public static class AbortPolicy implements RejectedExecutionHandler // 拋出java.util.concurrent.RejectedExecutionException異常 public static class CallerRunsPolicy implements RejectedExecutionHandler //直接在 execute 方法的調用線程中運行被拒絕的任務。如果執行程序已關閉，則會丟棄該任務 public static class DiscardPolicy implements RejectedExecutionHandler // 不做任何處理 直接丟棄 public static class DiscardOldestPolicy implements RejectedExecutionHandler // 丟棄老的
自定義拒絕策略：
new RejectedExecutionHandler() { // 自定義拒絕策略 @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { try { // 如果LinkedBlockingQueue存滿了，阻塞等待有空間後再加入元素。（put方法是阻塞的） LOGGER.info("LinkedBlockingQueue has been full "); // put() 方法是阻塞的，如果隊列沒有空間會一直等待。 executor.getQueue().put(r); LOGGER.info("thread has been put in"); } catch (InterruptedException e) { e.printStackTrace(); } }
總結一點：當用java內置的一些工具的時候，如果有不理解的一定要 深入去看源碼。從根本上找解決思路。
java內置線程池ThreadPoolExecutor源碼學習記錄