創建ThreadPoolExecutor可以通過構造方法和Executors的靜態方法。
構造方法:
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
corePoolSize,線程池里最小線程數
maximumPoolSize,線程池里最大線程數量,超過最大線程時候會使用RejectedExecutionHandler
keepAliveTime,unit,線程最大的存活時間
workerQueue,緩存異步任務的隊列
threadFactory,用來構造線程池里的worker線程
線程池提交任務流程,以下代碼有簡化
Runnable command
if (workerCount < corePoolSize) {
new Worker().run(command);
workerCount++
}
if (workerQueue.offer(command)) { //入隊
// 還沒達到maximumPoolSize 創建worker
new Worker();
workerCount++;
} else if (!addWorker(command)) { //還沒達到maximumPoolSize,創建worker並運行
handler.rejectedException(command)
}
線程池中線程不足corePoolSize的時候直接創建線程運行command,不然的話提交到隊列,其他worker會對這個隊列做poll。
當workerQueue隊列滿了的時候,會創建新的worker,如果worker達到上限會調用handler.rejectedException方法。
Executors.newCachedThreadPool
core是0,max是無限,隊列是SynchronousQueue。無限創建線程,線程60S過期被銷毀。風險點是線程數不可控。
Executors.newFixedThreadPool
core是thread數量,max是thread數量,隊列是LinkedBlockingQueue。因為LinkedBlockingQueue是無界的,所以max不會起作用。風險點是workerQueue的長度不可控
關於ThreadPoolExecutor.DiscardPolicy和ThreadPoolExecutor.CallerRunsPolicy
一個是隊列慢了就丟棄。一個是隊列滿了就使用主線程來執行,變成同步調用的方式。主線程就是調用submit方法的線程。
我的任務是請求外部HTTP鏈接,且允許失敗,我希望不要影響其他線程。
使用線程池的參數:
new ThreadPoolExecutor(
10, 100, 60, TimeUnit.SECONDS, //限制最大線程,如果是CPU密集型的任務,核心線程就用核數,最大線程就用核數的N倍
new ArrayBlockingQueue<Runnable>(1000), //固定大小,防止內存堆積過多
new NamedThreadFactory("third_retrieval"), //自己的ThreadFactory
new ThreadPoolExecutor.DiscardPolicy()) //當任務對接的時候直接丟棄