原文連結:https://mp.weixin.qq.com/s/tcvSzucqCZCCCeoPh5vVhQ
一、擴充套件ThreadPoolExecutor1、擴充套件方法介紹ThreadPoolExecutor是可以擴充套件的,它內部提供了幾個可以在子類中改寫的方法(紅框內)。JDK內的註解上說,這些方法可以用以新增日誌,計時、監視或進行統計資訊的收集。是不是感覺很熟悉?有沒有一種spring aop中 @Around @Before @After三個註解的既視感?
我們來對比一下
其實他們的效果是一樣的,只是一個線上程池裡,一個在攔截器中。
對於ThreadPoolExecutor中的這些方法,有這樣的一些特點:
1、無論任務是從run中正常返回,還是丟擲一個異常而返回,afterExecute都會被呼叫(但是如果任務在完成後帶有一個Error,那麼就不會呼叫afterExecute)2、同時,如果beforeExecute丟擲一個RuntimeExecption,那麼任務將不會被執行,連帶afterExecute也不會被呼叫了。3、線上程池完成關閉操作時會呼叫terminated,類似於try-catch中的finally操作一樣。terminated可以用來釋放Executor在其生命週期裡分配的各種資源,此外也可以用來執行傳送通知、記錄日誌亦或是收集finalize統計資訊等操作。2、擴充套件方法實現我們先構建一個自定義的執行緒池,它透過擴充套件方法來新增日誌記錄和統計資訊的收集。為了測量任務的執行時間,beforeExecute必須記錄開始時間並把它儲存到一個afterExecute可以訪問的地方,於是用ThreadLocal來儲存變數,用afterExecute來讀取,並透過terminated來輸出平均任務和日誌訊息。
public class WeedThreadPool extends ThreadPoolExecutor { private final ThreadLocal<Long> startTime =new ThreadLocal<>(); private final Logger log =Logger.getLogger("WeedThreadPool"); //統計執行次數 private final AtomicLong numTasks =new AtomicLong(); //統計總執行時間 private final AtomicLong totalTime =new AtomicLong(); /** * 這裡是實現執行緒池的構造方法,我隨便選了一個,大家可以根據自己的需求找到合適的構造方法 */ public WeedThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue); } //執行緒執行之前呼叫 protected void beforeExecute(Thread t,Runnable r){ super.beforeExecute(t,r); System.out.println(String.format("Thread %s:start %s",t,r)); //因為currentTimeMillis返回的是ms,而眾所周知ms是很難產生差異的,所以換成了nanoTime用ns來展示 startTime.set(System.nanoTime()); } //執行緒執行之後呼叫 protected void afterExecute(Runnable r,Throwable t){ try { Long endTime =System.nanoTime(); Long taskTime =endTime-startTime.get(); numTasks.incrementAndGet(); totalTime.addAndGet(taskTime); System.out.println(String.format("Thread %s:end %s, time=%dns",Thread.currentThread(),r,taskTime)); }finally { super.afterExecute(r,t); } } //執行緒池退出時候呼叫 protected void terminated(){ try{ System.out.println(String.format("Terminated: avg time =%dns, ",totalTime.get()/numTasks.get())); }finally { super.terminated(); } }}
測試案例:
public class WeedThreadTest { BlockingQueue<Runnable> taskQueue; final static WeedThreadPool weedThreadPool =new WeedThreadPool(3,10,1, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(100)); public static void main(String[] args) throws InterruptedException { for(int i=0;i<3;i++) { weedThreadPool.execute(WeedThreadTest::run); } Thread.sleep(2000L); weedThreadPool.shutdown(); } private static void run() { System.out.println("thread id is: " + Thread.currentThread().getId()); try { Thread.sleep(1024L); } catch (InterruptedException e) { e.printStackTrace(); } }}
3、使用場景用到這些方法的地方其實和用到Spring AOP中一些場景比較相似,主要在記錄跟蹤、最佳化等方面可以使用,如日誌記錄和統計資訊的收集、測量任務的執行時間,以及一些任務完成之後傳送通知、郵件、資訊之類的。
二、CompletionService操作非同步任務1、非同步方法的原理如果我們意外收穫了一大批待執行的任務(舉個例子,比如去呼叫各大旅遊軟體的出行機票資訊),為了提高任務的執行效率,我們可以使用執行緒池submit非同步計算任務,透過呼叫Future介面實現類的get方法獲取結果。
雖然使用了執行緒池會提高執行效率,但是呼叫Future介面實現類的get方法是阻塞的,也就是和當前這個Future關聯的任務全部執行完成的時候,get方法才返回結果,如果當前任務沒有執行完成,而有其它Future關聯的任務已經完成了,就會白白浪費很多等待的時間。
所以,有沒有這樣一個方法,遍歷的時候誰先執行完成就先獲取哪個結果?
沒錯,我們的ExecutorCompletionService就可以實現這樣的效果,它的內部有一個先進先出的阻塞佇列,用於儲存已經執行完成的Future,透過呼叫它的take方法或poll方法可以獲取到一個已經執行完成的Future,進而透過呼叫Future介面實現類的get方法獲取最終的結果。
邏輯圖如下:
ExecutorCompletionService實現了CompletionService介面,在CompletionService介面中定義瞭如下這些方法:
Futuresubmit(Callabletask):提交一個Callable型別任務,並返回該任務執行結果關聯的Future;Futuresubmit(Runnable task,V result):提交一個Runnable型別任務,並返回該任務執行結果關聯的Future;Futuretake():從內部阻塞佇列中獲取並移除第一個執行完成的任務,阻塞,直到有任務完成;Futurepoll():從內部阻塞佇列中獲取並移除第一個執行完成的任務,獲取不到則返回null,不阻塞;Futurepoll(long timeout, TimeUnit unit):從內部阻塞佇列中獲取並移除第一個執行完成的任務,阻塞時間為timeout,獲取不到則返回null;2、非同步方法的實現public class WeedExecutorServiceDemo { /** * 繼續用之前建好的執行緒池,只是調整一下池大小 */ BlockingQueue<Runnable> taskQueue; final static WeedThreadPool weedThreadPool = new WeedThreadPool(1, 5, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(100)); public static Random r = new Random(); public static void main(String[] args) throws InterruptedException, ExecutionException { CompletionService<Integer> cs = new ExecutorCompletionService<Integer>(weedThreadPool); for (int i = 0; i < 3; i++) { cs.submit(() -> { //獲取計算任務 int init = 0; for (int j = 0; j < 100; j++) { init += r.nextInt(); } Thread.sleep(1000L); return Integer.valueOf(init); }); } weedThreadPool.shutdown(); /** * 透過take方法獲取,阻塞,直到有任務完成 */ for (int i = 0; i < 3; i++) { Future<Integer> future = cs.take(); if (future != null) { System.out.println(future.get()); } } }}
呼叫結果如下
我們也可以透過poll方法來獲取。
/** * 透過poll方法獲取 */ for (int i = 0; i < 3; i++) { System.out.println(cs.poll(1200L,TimeUnit.MILLISECONDS).get()); }
結果自然是一樣的
如果把阻塞時間改小一些,目前的程式碼就會出問題
/** * 透過poll方法獲取 */ for (int i = 0; i < 3; i++) { System.out.println(cs.poll(800L,TimeUnit.MILLISECONDS).get()); }
同樣的,poll方法也可以用來打斷超時執行的業務,比如在poll超時的情況下,直接呼叫執行緒池的shutdownNow(),殘暴地關閉整個執行緒池。
for (int i = 0; i < 3; i++) { Future<Integer> poll = cs.poll(800L, TimeUnit.MILLISECONDS); if (poll==null){ System.out.println("執行結束"); weedThreadPool.shutdownNow(); } }
3、使用場景選擇怎麼樣的方法來非同步執行任務,什麼樣的方式來接收任務,也是需要根據實際情況來考慮的。
1.、需要批次提交非同步任務的時候建議你使用 CompletionService。CompletionService 將執行緒池 Executor 和阻塞佇列 BlockingQueue 的功能融合在了一起,能夠讓批次非同步任務的管理更簡單。2、讓非同步任務的執行結果有序化。先執行完的先進入阻塞佇列,利用這個特性,你可以輕鬆實現後續處理的有序性,避免無謂的等待。3、執行緒池隔離。CompletionService支援建立自己的執行緒池,這種隔離效能避免幾個特別耗時的任務拖垮整個應用的風險。