1 CompletionService介紹
CompletionService用于提交一組Callable任務(wù)，其take方法返回已完成的一個(gè)Callable任務(wù)對(duì)應(yīng)的Future對(duì)象。
如果你向Executor提交了一個(gè)批處理任務(wù)，并且希望在它們完成后獲得結(jié)果。為此你可以將每個(gè)任務(wù)的Future保存進(jìn)一個(gè)集合，然后循環(huán)這個(gè)集合調(diào)用Future的get()取出數(shù)據(jù)。幸運(yùn)的是CompletionService幫你做了這件事情。
CompletionService整合了Executor和BlockingQueue的功能。你可以將Callable任務(wù)提交給它去執(zhí)行，然后使用類似于隊(duì)列中的take和poll方法，在結(jié)果完整可用時(shí)獲得這個(gè)結(jié)果，像一個(gè)打包的Future。
CompletionService的take返回的future是哪個(gè)先完成就先返回哪一個(gè)，而不是根據(jù)提交順序。
2 CompletionService源碼分析
首先看一下 構(gòu)造方法：
|
1
2
3
4
5
6
7
8
|
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)造法方法主要初始化了一個(gè)阻塞隊(duì)列，用來存儲(chǔ)已完成的task任務(wù)。
然后看一下 completionService.submit 方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
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()方法。
來看一下該方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
public void run() { //判斷執(zhí)行狀態(tài)，保證callable任務(wù)只被運(yùn)行一次 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對(duì)象中的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é)果：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
/** * 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ù)跟進(jìn)finishCompletion()方法，在該方法中找到 done()方法：
protected void done() { completionQueue.add(task); }
可以看到該方法只做了一件事情，就是將執(zhí)行結(jié)束的task添加到了隊(duì)列中，只要隊(duì)列中有元素，我們調(diào)用take()方法時(shí)就可以獲得執(zhí)行的結(jié)果。
到這里就已經(jīng)清晰了，異步非阻塞獲取執(zhí)行結(jié)果的實(shí)現(xiàn)原理其實(shí)就是通過隊(duì)列來實(shí)現(xiàn)的，FutureTask將執(zhí)行結(jié)果放到隊(duì)列中，先進(jìn)先出，線程執(zhí)行結(jié)束的順序就是獲取結(jié)果的順序。
CompletionService實(shí)際上可以看做是Executor和BlockingQueue的結(jié)合體。CompletionService在接收到要執(zhí)行的任務(wù)時(shí)，通過類似BlockingQueue的put和take獲得任務(wù)執(zhí)行的結(jié)果。CompletionService的一個(gè)實(shí)現(xiàn)是ExecutorCompletionService，ExecutorCompletionService把具體的計(jì)算任務(wù)交給Executor完成。
在實(shí)現(xiàn)上，ExecutorCompletionService在構(gòu)造函數(shù)中會(huì)創(chuàng)建一個(gè)BlockingQueue（使用的基于鏈表的無界隊(duì)列LinkedBlockingQueue），該BlockingQueue的作用是保存Executor執(zhí)行的結(jié)果。當(dāng)計(jì)算完成時(shí)，調(diào)用FutureTask的done方法。當(dāng)提交一個(gè)任務(wù)到ExecutorCompletionService時(shí)，首先將任務(wù)包裝成QueueingFuture，它是FutureTask的一個(gè)子類，然后改寫FutureTask的done方法，之后把Executor執(zhí)行的計(jì)算結(jié)果放入BlockingQueue中。
QueueingFuture的源碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
|
/** * 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實(shí)現(xiàn)任務(wù)
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
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可以更精確和簡(jiǎn)便地完成異步任務(wù)的執(zhí)行
CompletionService的一個(gè)實(shí)現(xiàn)是ExecutorCompletionService，它是Executor和BlockingQueue功能的融合體，Executor完成計(jì)算任務(wù)，BlockingQueue負(fù)責(zé)保存異步任務(wù)的執(zhí)行結(jié)果
在執(zhí)行大量相互獨(dú)立和同構(gòu)的任務(wù)時(shí)，可以使用CompletionService
CompletionService可以為任務(wù)的執(zhí)行設(shè)置時(shí)限，主要是通過BlockingQueue的poll(long time,TimeUnit unit)為任務(wù)執(zhí)行結(jié)果的取得限制時(shí)間，如果沒有完成就取消任務(wù)
到此這篇關(guān)于Java多線程 CompletionService的文章就介紹到這了,更多相關(guān)Java多線程CompletionService內(nèi)容請(qǐng)搜索服務(wù)器之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持服務(wù)器之家！
原文鏈接：https://juejin.cn/post/7018423693793558558
