1 CompletionService介绍
CompletionService 用于提交一组 Callable 任务，其take方法返回已完成的一个 Callable 任务对应的 Future 对象。
如果你向 Executor 提交了一个批处理任务，并且希望在它们完成后获得结果。为此你可以将每个任务的 Future 保存进一个集合，然后循环这个集合调用 Future 的 get() 取出数据。幸运的是 CompletionService 帮你做了这件事情。
CompletionService 整合了 Executor 和 BlockingQueue 的功能。你可以将 Callable 任务提交给它去执行，然后使用类似于队列中的take和poll方法，在结果完整可用时获得这个结果，像一个打包的 Future 。
CompletionService 的take返回的 future 是哪个先完成就先返回哪一个，而不是根据提交顺序。
2 CompletionService源码分析
首先看一下 构造方法：
|
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 测试数据pletionQueue = new LinkedBlockingQueue<Future<V>>(); } |
构造法方法主要初始化了一个阻塞队列，用来存储已完成的 task 任务。
然后看一下 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 任务被包装成 QueueingFuture ，而 QueueingFuture 是 FutureTask 的子类，所以最终执行了 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() { //判断执行状态，保证callable任务只被运行一次 if (state != NEW || !UNSAFE测试数据pareAndSwapObject( this , runnerOffset, null , Thread.currentThread())) return ; try { Callable<V> c = callable; if (c != null && state == NEW) { V result; boolean ran; try { //这里回调我们创建的callable对象中的call方法 result = c.call(); ran = true ; } catch (Throwable ex) { result = null ; ran = false ; setException(ex); } if (ran) //处理执行结果 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 中执行 run 方法，最终回调自定义的 callable 中的 call 方法，执行结束之后，
通过 set(result) 处理执行结果：
|
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测试数据pareAndSwapInt( this , stateOffset, NEW, COMPLETING)) { outcome = v; UNSAFE.putOrderedInt( this , stateOffset, NORMAL); // final state finishCompletion(); } } |
继续跟进 finishCompletion() 方法，在该方法中找到 done() 方法：
protected void done() { completionQueue.add(task); }
可以看到该方法只做了一件事情，就是将执行结束的 task 添加到了队列中，只要队列中有元素，我们调用 take() 方法时就可以获得执行的结果。
到这里就已经清晰了，异步非阻塞获取执行结果的实现原理其实就是通过队列来实现的， FutureTask 将执行结果放到队列中，先进先出，线程执行结束的顺序就是获取结果的顺序。
CompletionService 实际上可以看做是 Executor 和 BlockingQueue 的结合体。 CompletionService 在接收到要执行的任务时，通过类似 BlockingQueue 的put和take获得任务执行的结果。 CompletionService 的一个实现是 ExecutorCompletionService ， ExecutorCompletionService 把具体的计算任务交给 Executor 完成。
在实现上， ExecutorCompletionService 在构造函数中会创建一个 BlockingQueue （使用的基于链表的无界队列LinkedBlockingQueue），该 BlockingQueue 的作用是保存 Executor 执行的结果。当计算完成时，调用 FutureTask 的done方法。当提交一个任务到 ExecutorCompletionService 时，首先将任务包装成 QueueingFuture ，它是 FutureTask 的一个子类，然后改写 FutureTask 的done方法，之后把 Executor 执行的计算结果放入 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实现任务
|
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总结
相比 ExecutorService ， CompletionService 可以更精确和简便地完成异步任务的执行
CompletionService 的一个实现是 ExecutorCompletionService ，它是 Executor 和 BlockingQueue 功能的融合体， Executor 完成计算任务， BlockingQueue 负责保存异步任务的执行结果
在执行大量相互独立和同构的任务时，可以使用 CompletionService
CompletionService 可以为任务的执行设置时限，主要是通过 BlockingQueue 的 poll (long time,TimeUnit unit)为任务执行结果的取得限制时间，如果没有完成就取消任务
到此这篇关于Java多线程 CompletionService 的文章就介绍到这了,更多相关Java多线程 CompletionService 内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！
原文链接：https://juejin.cn/post/7018423693793558558
查看更多关于Java多线程 CompletionService的详细内容...