CompletionService和ExecutorCompletionService详解

旧城等待， 2024-04-17 06:04 52阅读 0赞

1、CompletionService接口定义了一组任务管理接口:

*  submit() - 提交任务
 *  take() - 获取任务结果
 *  poll() - 获取任务结果

通常维护一个list保存submit的callable task所返回的Future对象，在主线程中遍历这个list并调用Future的get()方法取到Task的返回值

2、ExecutorCompletionService类是CompletionService接口的实现

*  ExecutorCompletionService内部管理者一个已完成任务的阻塞队列
 *  ExecutorCompletionService引用了一个Executor, 用来执行任务
 *  submit()方法最终会委托给内部的executor去执行任务
 *  take/poll方法的工作都委托给内部的已`完成任务阻塞队列`
 *  如果阻塞队列中有已完成的任务, take方法就返回任务的结果, 否则阻塞等待任务完成

3、poll与take方法不同, poll有两个版本:

*  无参的poll方法 --- 如果完成队列中有数据就返回, 否则返回null
 *  有参数的poll方法 --- 如果完成队列中有数据就直接返回, 否则等待指定的时间, 到时间后如果还是没有数据就返回null
 *  ExecutorCompletionService主要用与管理异步任务 (有结果的任务, 任务完成后要处理结果)

4、简单的使用例子

/**
         * <三> 使用ExecutorCompletionService管理异步任务
         * 1. Java中的ExecutorCompletionService<V>本身有管理任务队列的功能
         *    i. ExecutorCompletionService内部维护列一个队列, 用于管理已完成的任务
         *    ii. 内部还维护列一个Executor, 可以执行任务
         *
         * 2. ExecutorCompletionService内部维护了一个BlockingQueue, 只有完成的任务才被加入到队列中
         *
         * 3. 任务一完成就加入到内置管理队列中, 如果队列中的数据为空时, 调用take()就会阻塞 (等待任务完成)
         *    i. 关于完成任务是如何加入到完成队列中的, 请参考ExecutorCompletionService的内部类QueueingFuture的done()方法
         *
         * 4. ExecutorCompletionService的take/poll方法是对BlockingQueue对应的方法的封装, 关于BlockingQueue的take/poll方法:
         *    i. take()方法, 如果队列中有数据, 就返回数据, 否则就一直阻塞;
         *    ii. poll()方法: 如果有值就返回, 否则返回null
         *    iii. poll(long timeout, TimeUnit unit)方法: 如果有值就返回, 否则等待指定的时间; 如果时间到了如果有值, 就返回值, 否则返回null
         *
         * 解决了已完成任务得不到及时处理的问题
         */
        static void case3() throws InterruptedException, ExecutionException {
            Random random = new Random();
    
            ExecutorService service = Executors.newFixedThreadPool(10);
            ExecutorCompletionService<String> completionService = new ExecutorCompletionService<String>(service);
    
            for(int i=0; i<50; i++) {
                completionService.submit(new Callable<String>() {
                    @Override
                    public String call() throws Exception {
                        Thread.sleep(random.nextInt(5000));
                        return Thread.currentThread().getName();
                    }
                });
            }
    
            int completionTask = 0;
            while(completionTask < 50) {
                //如果完成队列中没有数据, 则阻塞; 否则返回队列中的数据
                Future<String> resultHolder = completionService.take();
                System.out.println("result: " + resultHolder.get());
                completionTask++;
            }
    
            System.out.println(completionTask + " task done !");
    
            //ExecutorService使用完一定要关闭 (回收资源, 否则系统资源耗尽! .... 呵呵...)
            service.shutdown();
        }

5、ExecutorCompletionService是如何执行任务, 又是如何将任务的结果存储到完成队列中的

*  ExecutorCompletionService在submit任务时, 会创建一个QueueingFuture, 然后将创建的QueueingFuture丢给executor, 让executor完成任务的执行工作
 *  QueueingFuture继承与FutureTask类, 而FutureTask实现了两个接口Runnable和Future
 *  Runnable一般表示要执行的任务的过程, 而Future则表述执行任务的 结果 (或者说是任务的一个句柄, 可获取结果, 取消任务等)
 *  因此FutureTask就是一个有结果可期待的任务. FutureTask实现了run方法, 我们指定此方法一般是在在工作线程(不是submit线程) 执行的