?【Java技术指南】「并发编程专题」CompletionService框架基本使用和原理探究(基础篇)

时间:2021-09-13 17:24:37   收藏:0   阅读:34

前提概要

在开发过程中在使用多线程进行并行处理一些事情的时候,大部分场景在处理多线程并行执行任务的时候,可以通过List添加Future来获取执行结果,有时候我们是不需要获取任务的执行结果的,方便后面引出ExecutorCompletionService。

CompletionService的介绍

CompletionService的实现原理

public interface CompletionService<V> {
    Future<V> submit(Callable<V> task);
    Future<V> submit(Runnable task, V result);
    Future<V> take() throws InterruptedException;
    Future<V> poll();
    Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
}

方法摘要

提交一个 Callable 任务;一旦完成,便可以由take()、poll()方法获取

Future submit(Callable task):

提交一个 Runnable 任务,并指定计算结果;

Future submit(Runnable task, V result):

获取并移除表示下一个已完成任务的 Future,如果目前不存在这样的任务,则等待。

Future take() throws InterruptedException

获取并移除表示下一个已完成任务的 Future,如果不存在这样的任务,则返回 null。

Future poll()

获取并移除表示下一个已完成任务的 Future,如果目前不存在这样的任务,则将等待指定的时间(如果有必要)。

Future poll(long timeout, TimeUnit unit) throws InterruptedException

例子,程序提交了多个任务,但只要有一个任务完成并返回一个非空的结果,并可以忽略掉其余的任务。

 void eample(Executor e, Collection<Callable<Result>> solvers) throws InterruptedException {
     CompletionService<Result> completionService = new ExecutorCompletionService<Result>(e);
     int n = solvers.size();
     List<Future<Result>> futures = new ArrayList<Future<Result>>(n);
     Result result = null;
     try {
         //提交多个任务
         for (Callable<Result> s : solvers)
             futures.add(completionService.submit(s));
        //
         for (int i = 0; i < n; ++i) {
             try {
                 //等待获取一个已经完成的任务
                 Result r = completionService.take().get();
                 //判断返回结果是否为空
                 if (r != null) {
                     result = r;
                     break;
                 }
             } catch (ExecutionException ignore) {}
         }
     }
     finally {
         //取消所有任务
         for (Future<Result> f : futures)
               f.cancel(true);
         }
     if (result != null)
         use(result);
 }

ExecutorCompletionService的介绍

构造方法

ExecutorCompletionService(Executor executor):

指定了任务执行器Executor和已完成的任务队列completionQueue

ExecutorCompletionService(Executor executor, BlockingQueue<Future> completionQueue)
实现构造器
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>>();
}
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;
}
QueueingFuture为内部类:

在提交任务时,将任务封装成QueueingFuture:

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;
}

其中,done()方法就是在任务执行完毕后,将任务放入队列中。

最后,如果工作中并行处理任务不需要获取结果的,我们正常使用线程池提交就可以,任务技术只要适合工作的业务场景就是好的。

原文:https://www.cnblogs.com/liboware/p/15259870.html

评论(0
© 2014 bubuko.com 版权所有 - 联系我们:wmxa8@hotmail.com
打开技术之扣,分享程序人生!