Java中的CompletionService批量異步執(zhí)行詳解
前景引入
我們知道線程池可以執(zhí)行異步任務(wù)，同時可以通過返回值Future獲取返回值，所以異步任務(wù)大多數(shù)采用ThreadPoolExecutor+Future，如果存在如下情況，需要從任務(wù)一二三中獲取返回值后，保存到數(shù)據(jù)庫中，用異步邏輯實現(xiàn)代碼應(yīng)該如下所示。
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(3);
Future<Integer> f1 = executorService.submit(() -> {
System.out.println("執(zhí)行任務(wù)一");
return 1;
});
Future<Integer> f2 = executorService.submit(() -> {
System.out.println("執(zhí)行任務(wù)二");
return 2;
});
Future<Integer> f3 = executorService.submit(() -> {
System.out.println("執(zhí)行任務(wù)三");
return 3;
});
Integer r1 = f1.get();
executorService.execute(()->{
// 省略保存r1操作
System.out.println(r1);
});
Integer r2 = f2.get();
executorService.execute(()->{
// 省略保存r2操作
System.out.println(r2);
});
Integer r3 = f3.get();
executorService.execute(()->{
// 省略保存r3操作
System.out.println(r3);
});
executorService.shutdown();
}這樣寫的代碼一點毛病沒有，邏輯都是正常的，但如果存在任務(wù)一查詢了比較耗時的操作，由于f1.get是阻塞執(zhí)行，那么就算任務(wù)二和任務(wù)三已經(jīng)返回結(jié)果，任務(wù)二的返回值和任務(wù)三的返回值都是不能保存到數(shù)據(jù)庫的，因為f1.get將主線程阻塞了。
批量異步實現(xiàn)
那可以如何處理呢？可以采用萬能的阻塞隊列，任務(wù)先執(zhí)行完畢的先入隊，這樣可以保證其它線程入庫的速度不受影響，提高效率。
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(3);
ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue(3);
Future<Integer> f1 = executorService.submit(() -> {
System.out.println("執(zhí)行任務(wù)一");
Thread.sleep(5000);
return 1;
});
Future<Integer> f2 = executorService.submit(() -> {
System.out.println("執(zhí)行任務(wù)二");
return 2;
});
Future<Integer> f3 = executorService.submit(() -> {
System.out.println("執(zhí)行任務(wù)三");
Thread.sleep(3000);
return 3;
});
executorService.execute(()->{
try {
Integer r1 = f1.get();
// 阻塞隊列入隊操作
queue.put(r1);
System.out.println(r1);
} catch (Exception e) {
e.printStackTrace();
}
});
executorService.execute(()->{
try {
Integer r2 = f2.get();
queue.put(r2);
System.out.println(r2);
} catch (Exception e) {
e.printStackTrace();
}
});
executorService.execute(()->{
try {
Integer r3 = f3.get();
queue.put(r3);
System.out.println(r3);
} catch (Exception e) {
e.printStackTrace();
}
});
// 循環(huán)次數(shù)不要使用queue.size限制，因為不同時刻queue.size值是有可能不同的
for (int i = 0; i <3; i++) {
Integer integer = queue.take();
// 省略保存integer操作
executorService.execute(()->{
System.out.println("保存入庫=="+integer);
});
}
executorService.shutdown();
}產(chǎn)生結(jié)果如下
同樣的在生產(chǎn)中不建議使用，因為SDK為我們提供了工具類CompletionService，CompletionService內(nèi)部就維護了一個阻塞隊列，唯一與上述代碼實現(xiàn)有所區(qū)別的是，阻塞隊列入庫的是Future對象，其余原理類似。
CompletionService
如何創(chuàng)建CompletionService
CompletionService同樣是一個接口，其具體實現(xiàn)為ExecutorCompletionService，創(chuàng)建CompletionService對象有兩種方式
public ExecutorCompletionService(Executor executor); public ExecutorCompletionService(Executor executor,BlockingQueue<Future<V>> completionQueue)
CompletionService對象的創(chuàng)建都是需要指定線程池，如果在創(chuàng)建時沒有傳入阻塞對象，那么會采用默認(rèn)的LinkedBlockingQueue無界阻塞隊列，如果應(yīng)用到生產(chǎn)可能會產(chǎn)生OOM的情況，這是需要注意的。
CompletionService初體驗
CompletionService如何做到批量執(zhí)行異步任務(wù)呢，將上述場景采用CompletionService實現(xiàn)下
public static void main(String[] args) throws InterruptedException, ExecutionException {
ExecutorService executorService = Executors.newFixedThreadPool(3);
CompletionService completionService = new ExecutorCompletionService(executorService);
Future<Integer> f1 = completionService.submit(() -> {
System.out.println("執(zhí)行任務(wù)一");
Thread.sleep(5000);
return 1;
});
Future<Integer> f2 = completionService.submit(() -> {
System.out.println("執(zhí)行任務(wù)二");
return 2;
});
Future<Integer> f3 = completionService.submit(() -> {
System.out.println("執(zhí)行任務(wù)三");
Thread.sleep(3000);
return 3;
});
for (int i = 0; i <3 ; i++) {
Future take = completionService.take();
Integer integer = (Integer) take.get();
executorService.execute(()->{
System.out.println("執(zhí)行入庫=="+integer);
});
}
executorService.shutdown();
}CompletionService接口說明
CompletionService的方法不多，使用起來比較簡單，方法簽名如下
public static void main(String[] args) throws InterruptedException, ExecutionException {
ExecutorService executorService = Executors.newFixedThreadPool(3);
CompletionService completionService = new ExecutorCompletionService(executorService);
Future<Integer> f1 = completionService.submit(() -> {
System.out.println("執(zhí)行任務(wù)一");
Thread.sleep(5000);
return 1;
});
Future<Integer> f2 = completionService.submit(() -> {
System.out.println("執(zhí)行任務(wù)二");
return 2;
});
Future<Integer> f3 = completionService.submit(() -> {
System.out.println("執(zhí)行任務(wù)三");
Thread.sleep(3000);
return 3;
});
for (int i = 0; i <3 ; i++) {
Future take = completionService.take();
Integer integer = (Integer) take.get();
executorService.execute(()->{
System.out.println("執(zhí)行入庫=="+integer);
});
}
executorService.shutdown();
}總結(jié)
CompletionService主要是去解決無效等待的問題，如果一個耗時較長的任務(wù)在執(zhí)行，那么可以采用這種方式避免無效的等待
CompletionService還能讓異步任務(wù)的執(zhí)行結(jié)果有序化，先執(zhí)行完就先進入阻塞隊列。
到此這篇關(guān)于Java中的CompletionService批量異步執(zhí)行詳解的文章就介紹到這了,更多相關(guān)CompletionService批量異步執(zhí)行內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
java實現(xiàn)ReadWriteLock讀寫鎖的示例
ReadWriteLock是Java并發(fā)包中的接口,定義了讀鎖和寫鎖,讀鎖允許多線程同時訪問共享資源,而寫鎖則要求獨占,這種機制適用于讀多寫少的場景,可以提高并發(fā)效率同時保證數(shù)據(jù)一致性,本文就來詳細的介紹一下如何實現(xiàn),感興趣的可以了解一下2024-09-09
這篇文章主要為大家詳細介紹一下Java集合類中List的用法，文中的示例代碼講解詳細，對我們學(xué)習(xí)Java有一定幫助，感興趣的可以了解一下2022-07-07
這篇文章主要為大家詳細介紹了java多線程實現(xiàn)下載圖片并壓縮，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-05-05
Java實現(xiàn)統(tǒng)計文件夾下所有文件的字?jǐn)?shù)
這篇文章主要為大家詳細介紹了如何使用Java實現(xiàn)統(tǒng)計文件夾下所有文件的字?jǐn)?shù),文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2024-03-03
springboot基于docsify?實現(xiàn)隨身文檔
這篇文章主要介紹了springboot基于docsify實現(xiàn)隨身文檔的相關(guān)資料,需要的朋友可以參考下2022-09-09
淺談MyBatis3 DynamicSql風(fēng)格語法使用指南
這篇文章主要介紹了淺談MyBatis3 DynamicSql風(fēng)格語法使用指南，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03
這篇文章主要給大家介紹了如何優(yōu)雅的進行Spring整合MongoDB的相關(guān)資料，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧。2018-02-02
這篇文章主要介紹了Java獲取任意http網(wǎng)頁源代碼的方法,可實現(xiàn)獲取網(wǎng)頁代碼以及去除HTML標(biāo)簽的代碼功能,涉及Java正則操作相關(guān)實現(xiàn)技巧,需要的朋友可以參考下2017-09-09
Java BeanMap實現(xiàn)Bean與Map的相互轉(zhuǎn)換
這篇文章主要介紹了利用BeanMap進行對象與Map的相互轉(zhuǎn)換,通過net.sf.cglib.beans.BeanMap類中的方法來轉(zhuǎn)換，效率極高，本文給大家分享實現(xiàn)代碼，感興趣的朋友一起看看吧2022-11-11
這篇文章主要介紹了java中的GC收集器，GC（Garbage collection ）指的是程序內(nèi)存管理分手動和自動，手動內(nèi)存管理，需要我們編程的時候顯式分配和釋放空間，但如果忘記釋放，會造成嚴(yán)重的內(nèi)存泄漏問題，下面文章內(nèi)容我們就來實例說明情況,需要的朋友可以參考一下2021-10-10