【并发编程】Java5 - CompletionService，将异步执行与获取结果分离

一时失言乱红尘 2023-10-08 22:41 17阅读 0赞

1. 简介

相比Future([【并发编程】Java5 - Future，基本使用][Java5 - Future])，CompletionService除了支持并行执行任务并获取结果外，还支持优先获取到最快执行的任务结果，但CompletionService要求并行执行的任务是无序的。  
  使用Future的实现类FutureTask获取多个任务的结果时，当任务未执行完成，主线程会阻塞，只到任务执行完成才会获取到结果；CompletionService的实现类ExecutorCompletionService内部维护了一个存储Future的BlockingQueue，任务执行完成后会把Future保存到队列中。当获取多个任务的结果时，会从BlockingQueue中调用take()方法获取到Future再调用get()方法获取结果，如果暂时没有任务执行完成，则阻塞直到有任务执行完成并保存到BlockingQueue中，这样获取到结果的顺序会按照任务执行快慢的顺序依次返回。

2. CompletionService接口API

<table> 
 <thead> 
  <tr> 
   <th>方法及参数</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Future submit(Callable task)</td> 
   <td>提交实现Callable的线程任务</td> 
  </tr> 
  <tr> 
   <td>Future submit(Runnable task, V result)</td> 
   <td>提交实现Runnable的线程任务，并指定预期结果</td> 
  </tr> 
  <tr> 
   <td>Future take()</td> 
   <td>获取已完成任务的 Future，没有则阻塞</td> 
  </tr> 
  <tr> 
   <td>Future poll()</td> 
   <td>获取已完成任务的 Future，没有则返回null</td> 
  </tr> 
  <tr> 
   <td>Future poll(long timeout, TimeUnit unit)</td> 
   <td>在指定时间内获取已完成任务的 Future，没有则返回null</td> 
  </tr> 
 </tbody> 
</table>

3. CompletionService使用场景

*  获取到任务结果后还需要继续处理，如果获取到结果后立即返回，则无法提现出CompletionService的优势
 *  多个任务并发执行，只需要获取到任意一个任务的结果即可

4. CompletionService应用

*  前期准备：实现Callable创建线程，支持指定执行时间

/**
     * 创建线程
     */
    @RequiredArgsConstructor
    class MyCallable implements Callable<Long> {
    
        /**
         * 任务执行时间
         */
        private final long execTime;
    
        @Override
        public Long call() throws Exception {
            Thread.sleep(execTime);
            return execTime;
        }
    }

*  Future：测试并行执行多个任务获取结果后继续业务处理

/**
     * Future：测试并行执行多个任务获取结果后继续业务处理
     *
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @Test
    public void testFuture() throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
    
        List<Future<Long>> futureList = new ArrayList<>(3);
    
        // 记录开始时间
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
    
        // 创建5个任务
        futureList.add(executorService.submit(new MyCallable(5000)));
        futureList.add(executorService.submit(new MyCallable(4000)));
        futureList.add(executorService.submit(new MyCallable(3000)));
        futureList.add(executorService.submit(new MyCallable(2000)));
        futureList.add(executorService.submit(new MyCallable(1000)));
    
        // 阻塞获取结果
        for (Future<Long> future : futureList) {
            Long result = future.get();
            log.debug("执行结果：" + result);
    
            // 模拟获取结果后业务处理
            Thread.sleep(500);
        }
    
        stopWatch.stop();
        log.info("总耗时：" + stopWatch.getTotalTimeMillis() + " ms");
    
        executorService.shutdown();
    }

控制台打印：

18:53:06.185 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：5000
    18:53:06.690 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：4000
    18:53:07.190 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：3000
    18:53:07.690 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：2000
    18:53:08.191 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：1000
    18:53:08.691 [main] INFO com.c3stones.test.CompletionServiceTest - 总耗时：7516 ms

通过结果可以看出，结果按照提交的顺序依次返回，总耗时 = 每个任务执行时间 + 获取结果后业务处理时间。

*  CompletionService：测试并行执行多个任务获取结果后继续业务处理

/**
     * CompletionService：测试并行执行多个任务获取结果后继续业务处理
     *
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @Test
    public void testCompletionService() throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        CompletionService<Long> completionService = new ExecutorCompletionService<>(executorService);
    
        // 记录开始时间
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
    
        // 创建5个任务
        completionService.submit(new MyCallable(5000));
        completionService.submit(new MyCallable(4000));
        completionService.submit(new MyCallable(3000));
        completionService.submit(new MyCallable(2000));
        completionService.submit(new MyCallable(1000));
    
        for (int i = 0; i < 5; i++) {
            Long result = completionService.take().get();
            log.debug("执行结果：" + result);
    
            // 模拟获取结果后业务处理
            Thread.sleep(500);
        }
    
        stopWatch.stop();
        log.info("总耗时：" + stopWatch.getTotalTimeMillis() + " ms");
    
        executorService.shutdown();
    }

控制台打印：

18:55:25.727 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：1000
    18:55:26.724 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：2000
    18:55:27.725 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：3000
    18:55:28.725 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：4000
    18:55:29.724 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：5000
    18:55:30.225 [main] INFO com.c3stones.test.CompletionServiceTest - 总耗时：5505 ms

通过结果可以看出，结果按照提交的任务执行快慢顺序依次返回，总耗时 = 最长任务处理时间 + 获取结果后业务处理时间。

*  CompletionService：测试快速获取任意一个结果

/**
     * CompletionService：测试快速获取任意一个结果
     */
    @Test
    public void testFastResult() throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        CompletionService<Long> completionService = new ExecutorCompletionService<>(executorService);
    
        // 记录开始时间
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
    
        // 创建3个任务
        completionService.submit(new MyCallable(3000));
        completionService.submit(new MyCallable(2000));
        completionService.submit(new MyCallable(1000));
    
        Long result = completionService.take().get();
        log.debug("执行结果：" + result);
    
        stopWatch.stop();
        log.info("总耗时：" + stopWatch.getTotalTimeMillis() + " ms");
    
        executorService.shutdown();
    }

控制台打印：

18:59:58.104 [main] DEBUG com.c3stones.test.CompletionServiceTest - 执行结果：1000
    18:59:58.112 [main] INFO com.c3stones.test.CompletionServiceTest - 总耗时：1126 ms

通过结果可以看出，最快的任务执行完成后结束。

5. 项目地址

[thread-demo][]

[Java5 - Future]: https://www.cnblogs.com/cao-lei/p/17241139.html
[thread-demo]: https://github.com/C3Stones/blog