Java多線程 CompletionService
1 CompletionService介紹
CompletionService用于提交一組Callable任務(wù),其take方法返回已完成的一個Callable任務(wù)對應(yīng)的Future對象。
如果你向Executor提交了一個批處理任務(wù),并且希望在它們完成后獲得結(jié)果。為此你可以將每個任務(wù)的Future保存進一個集合,然后循環(huán)這個集合調(diào)用Future的get()取出數(shù)據(jù)。幸運的是CompletionService幫你做了這件事情。
CompletionService整合了Executor和BlockingQueue的功能。你可以將Callable任務(wù)提交給它去執(zhí)行,然后使用類似于隊列中的take和poll方法,在結(jié)果完整可用時獲得這個結(jié)果,像一個打包的Future。
CompletionService的take返回的future是哪個先完成就先返回哪一個,而不是根據(jù)提交順序。
2 CompletionService源碼分析
首先看一下 構(gòu)造方法:
public ExecutorCompletionService(Executor executor) {
if (executor == null)
throw new NullPointerException();
this.executor = executor;
this.aes = (executor instanceof AbstractExecutorService) ?
(AbstractExecutorService) executor : null;
this.completionQueue = new LinkedBlockingQueue<Future<V>>();
}
構(gòu)造法方法主要初始化了一個阻塞隊列,用來存儲已完成的task任務(wù)。
然后看一下 completionService.submit 方法:
public Future<V> submit(Callable<V> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task);
executor.execute(new QueueingFuture(f));
return f;
}
public Future<V> submit(Runnable task, V result) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task, result);
executor.execute(new QueueingFuture(f));
return f;
}
可以看到,callable任務(wù)被包裝成QueueingFuture,而 QueueingFuture是 FutureTask的子類,所以最終執(zhí)行了FutureTask中的run()方法。
來看一下該方法:
public void run() {
//判斷執(zhí)行狀態(tài),保證callable任務(wù)只被運行一次
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
//這里回調(diào)我們創(chuàng)建的callable對象中的call方法
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
//處理執(zhí)行結(jié)果
set(result);
}
} finally {
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
可以看到在該 FutureTask 中執(zhí)行run方法,最終回調(diào)自定義的callable中的call方法,執(zhí)行結(jié)束之后,
通過 set(result) 處理執(zhí)行結(jié)果:
/**
* Sets the result of this future to the given value unless
* this future has already been set or has been cancelled.
*
* <p>This method is invoked internally by the {@link #run} method
* upon successful completion of the computation.
*
* @param v the value
*/
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
繼續(xù)跟進finishCompletion()方法,在該方法中找到 done()方法:
protected void done() { completionQueue.add(task); }
可以看到該方法只做了一件事情,就是將執(zhí)行結(jié)束的task添加到了隊列中,只要隊列中有元素,我們調(diào)用take()方法時就可以獲得執(zhí)行的結(jié)果。
到這里就已經(jīng)清晰了,異步非阻塞獲取執(zhí)行結(jié)果的實現(xiàn)原理其實就是通過隊列來實現(xiàn)的,FutureTask將執(zhí)行結(jié)果放到隊列中,先進先出,線程執(zhí)行結(jié)束的順序就是獲取結(jié)果的順序。
CompletionService實際上可以看做是Executor和BlockingQueue的結(jié)合體。CompletionService在接收到要執(zhí)行的任務(wù)時,通過類似BlockingQueue的put和take獲得任務(wù)執(zhí)行的結(jié)果。CompletionService的一個實現(xiàn)是ExecutorCompletionService,ExecutorCompletionService把具體的計算任務(wù)交給Executor完成。
在實現(xiàn)上,ExecutorCompletionService在構(gòu)造函數(shù)中會創(chuàng)建一個BlockingQueue(使用的基于鏈表的無界隊列LinkedBlockingQueue),該BlockingQueue的作用是保存Executor執(zhí)行的結(jié)果。當(dāng)計算完成時,調(diào)用FutureTask的done方法。當(dāng)提交一個任務(wù)到ExecutorCompletionService時,首先將任務(wù)包裝成QueueingFuture,它是FutureTask的一個子類,然后改寫FutureTask的done方法,之后把Executor執(zhí)行的計算結(jié)果放入BlockingQueue中。
QueueingFuture的源碼如下:
/**
* FutureTask extension to enqueue upon completion
*/
private class QueueingFuture extends FutureTask<Void> {
QueueingFuture(RunnableFuture<V> task) {
super(task, null);
this.task = task;
}
protected void done() { completionQueue.add(task); }
private final Future<V> task;
}
3 CompletionService實現(xiàn)任務(wù)
public class CompletionServiceTest {
public static void main(String[] args) {
ExecutorService threadPool = Executors.newFixedThreadPool(10);
CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(threadPool);
for (int i = 1; i <=10; i++) {
final int seq = i;
completionService.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
Thread.sleep(new Random().nextInt(5000));
return seq;
}
});
}
threadPool.shutdown();
for (int i = 0; i < 10; i++) {
try {
System.out.println(
completionService.take().get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
}
7
3
9
8
1
2
4
6
5
10
4 CompletionService總結(jié)
相比ExecutorService,CompletionService可以更精確和簡便地完成異步任務(wù)的執(zhí)行
CompletionService的一個實現(xiàn)是ExecutorCompletionService,它是Executor和BlockingQueue功能的融合體,Executor完成計算任務(wù),BlockingQueue負責(zé)保存異步任務(wù)的執(zhí)行結(jié)果
在執(zhí)行大量相互獨立和同構(gòu)的任務(wù)時,可以使用CompletionService
CompletionService可以為任務(wù)的執(zhí)行設(shè)置時限,主要是通過BlockingQueue的poll(long time,TimeUnit unit)為任務(wù)執(zhí)行結(jié)果的取得限制時間,如果沒有完成就取消任務(wù)
到此這篇關(guān)于Java多線程 CompletionService的文章就介紹到這了,更多相關(guān)Java多線程CompletionService內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
java自動生成編號的實現(xiàn)(格式:yyMM+四位流水號)
這篇文章主要介紹了java自動生成編號的實現(xiàn)(格式:yyMM+四位流水號),文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-10-10
SpringBoot整合JavaMail實現(xiàn)發(fā)郵件的項目實踐
本文主要介紹了SpringBoot整合JavaMail實現(xiàn)發(fā)郵件的項目實踐,詳細闡述了使用SpringBoot和JavaMail發(fā)送郵件的步驟,具有一定的參考價值,感興趣的可以了解一下2023-10-10
如何在Spring?Boot中使用MyBatis訪問數(shù)據(jù)庫
MyBatis可以通過簡單的XML或者注解來配置和映射原始類型,接口,和Java POJO為數(shù)據(jù)庫中記錄,使用MyBatis幫助我們解決各種問題,本文介紹如何在Spring?Boot中使用MyBatis訪問數(shù)據(jù)庫,感興趣的朋友一起看看吧2023-11-11

