Java 并发编程 Future及CompletionService

Future

Future用于异步结果计算。它提供了一些方法来检查计算是否完成，使用get方法将阻塞线程直到结果返回

• cancel：尝试取消任务的执行，如果任务已完成或已取消，此操作无效
• isCancelled：任务是否已取消
• isDone：任务是否已完成
• get：阻塞线程以获取计算结果，直至任务执行完毕返回结果
• get(long timeout, TimeUnit unit)：阻塞线程以获取计算结果，若在指定时间没返回结果，则返回null

public  interface  Future<V> {
   
   
    boolean cancel(boolean  mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

Future结合线程池的使用

public void futureTest(){
   
   
    ExecutorService executorService = Executors.newFixedThreadPool(10);

    Future<String> nickFuture = executorService.submit(() -> userService.getNick());
    Future<String> nameFuture = executorService.submit(() -> userService.getUserName());

    // 阻塞开始，等待结果
    String nick = nickFuture.get(1000, TimeUnit.MILLISECONDS);
    String name = nameFuture.get();
}

CompletionService

CompletionService整合了Executor和BlockingQueue的功能。将Callable任务提交给它去执行，使用take()和poll()获取最新完成的任务执行结果.

ExecutorCompletionService是该接口的实现类，内部有一个线程池和BlockingQueue队列。它的实现原理其实挺简单：每个提交给ExecutorCompletionService的任务，都会被封装成一个QueueingFuture（FutureTask的子类），它重写了done()方法（该方法会在任务执行完成之后回调），将执行完成的FutureTask加入到内部队列，take()和poll()将得到最新完成的结果FutrueTask。

使用方法

注：若线程池为局部变量，使用完后需关闭线程池。或直接使用全局变量命名的线程池

public Map<Long, List<String>> batchDoGetLiveRecord(List<LiveRecordReqDTO> reqList) {
   
   
    Map<Long, List<String>> recordMap = Maps.newHashMap();
    // 用线程池executor新建一个CompletionService
    CompletionService<Map<Long, List<String>>> completionService = new ExecutorCompletionService<>(executor);

    try {
   
   
        for (LiveRecordReqDTO req : reqList) {
   
   
            completionService.submit(() -> doGetLiveRecordAsMap(req));
        }

        for (int i = 0; i < reqList.size(); i++){
   
   
            Map<Long, List<String>> map = completionService.take().get(1000, TimeUnit.MILLISECONDS);
            recordMap.putAll(map);
        }
    } catch (Exception e) {
   
   
        logger.warn("batchDoGetLiveRecord fail, uidList:{}, bizId:{}",reqList, bizId, e);
    } finally {
   
   
        completionService.shutdown();
    }

    return recordMap;
}

类图自CompletionService和ExecutorCompletionService详解

参考资料：

1. CompletionService和ExecutorCompletionService详解
2. 并发处理利器-CompletionService