[Java并发-16] CompletionService：批量执行异步任务

咱们思考下这个场景：从三个电商询价，而后保存在本身的数据库里。经过以前所学，咱们可能这么实现。

// 建立线程池
ExecutorService executor =
  Executors.newFixedThreadPool(3);
// 异步向电商 S1 询价
Future<Integer> f1 = 
  executor.submit(
    ()->getPriceByS1());
// 异步向电商 S2 询价
Future<Integer> f2 = 
  executor.submit(
    ()->getPriceByS2());
// 异步向电商 S3 询价
Future<Integer> f3 = 
  executor.submit(
    ()->getPriceByS3());
    
// 获取电商 S1 报价并保存
r=f1.get();
executor.execute(()->save(r));
  
// 获取电商 S2 报价并保存
r=f2.get();
executor.execute(()->save(r));
  
// 获取电商 S3 报价并保存  
r=f3.get();
executor.execute(()->save(r));

上面的这个方案自己没有太大问题，可是有个地方的处理须要你注意，那就是若是获取电商 S1 报价的耗时很长，那么即使获取电商 S2 报价的耗时很短，也没法让保存 S2 报价的操做先执行，由于这个主线程都阻塞在了 f1.get(),那咱们如何解决了？

咱们能够增长一个阻塞队列，获取到 S一、S二、S3 的报价都进入阻塞队列，而后在主线程中消费阻塞队列，这样就能保证先获取到的报价先保存到数据库了。下面的示例代码展现了如何利用阻塞队列实现先获取到的报价先保存到数据库。

// 建立阻塞队列
BlockingQueue<Integer> bq =
  new LinkedBlockingQueue<>();
// 电商 S1 报价异步进入阻塞队列  
executor.execute(()->
  bq.put(f1.get()));
// 电商 S2 报价异步进入阻塞队列  
executor.execute(()->
  bq.put(f2.get()));
// 电商 S3 报价异步进入阻塞队列  
executor.execute(()->
  bq.put(f3.get()));
// 异步保存全部报价  
for (int i=0; i<3; i++) {
  Integer r = bq.take();
  executor.execute(()->save(r));
}

利用 CompletionService 实现询价系统

不过在实际项目中，并不建议你这样作，由于 Java SDK 并发包里已经提供了设计精良的 CompletionService。利用 CompletionService 能让代码更简练。

CompletionService 的实现原理也是内部维护了一个阻塞队列，当任务执行结束就把任务的执行结果加入到阻塞队列中，不一样的是 CompletionService 是把任务执行结果的 Future 对象加入到阻塞队列中，而上面的示例代码是把任务最终的执行结果放入了阻塞队列中。

那到底该如何建立 CompletionService 呢？

CompletionService 接口的实现类是 ExecutorCompletionService，这个实现类的构造方法有两个，分别是：

1. ExecutorCompletionService(Executor executor)
2. ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)

这两个构造方法都须要传入一个线程池，若是不指定 completionQueue，那么默认会使用无界的 LinkedBlockingQueue。任务执行结果的 Future 对象就是加入到 completionQueue 中。

下面的示例代码完整地展现了如何利用 CompletionService 来实现高性能的询价系统。其中，咱们没有指定 completionQueue，以后经过 CompletionService 接口提供的 submit() 方法提交了三个询价操做，这三个询价操做将会被 CompletionService 异步执行。最后，咱们经过 CompletionService 接口提供的 take() 方法获取一个 Future 对象，调用 Future 对象的 get() 方法就能返回询价操做的执行结果了。

// 建立线程池
ExecutorService executor = 
  Executors.newFixedThreadPool(3);
// 建立 CompletionService
CompletionService<Integer> cs = new 
  ExecutorCompletionService<>(executor);
// 异步向电商 S1 询价
cs.submit(()->getPriceByS1());
// 异步向电商 S2 询价
cs.submit(()->getPriceByS2());
// 异步向电商 S3 询价
cs.submit(()->getPriceByS3());
// 将询价结果异步保存到数据库
for (int i=0; i<3; i++) {
  Integer r = cs.take().get();
  executor.execute(()->save(r));
}

CompletionService 接口说明

下面咱们详细地介绍一下 CompletionService 接口提供的方法，CompletionService 接口提供的方法有 5 个，这 5 个方法的方法签名以下所示。

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;

CompletionService 后3 个方法，都是和阻塞队列相关的，take()、poll() 都是从阻塞队列中获取并移除一个元素；它们的区别在于若是阻塞队列是空的，那么调用 take() 方法的线程会被阻塞，而 poll() 方法会返回 null 值。

利用 CompletionService 实现 Dubbo 中的 Forking Cluster

Dubbo 中有一种叫作Forking 的集群模式，这种集群模式下，支持并行地调用多个查询服务，只要有一个成功返回结果，整个服务就能够返回了。例如你须要提供一个地址转坐标的服务，为了保证该服务的高可用和性能，你能够并行地调用 3 个地图服务商的 API，而后只要有 1 个正确返回告终果 r，那么地址转坐标这个服务就能够直接返回 r 了。这种集群模式能够容忍 2 个地图服务商服务异常，但缺点是消耗的资源偏多。

geocoder(addr) {
  // 并行执行如下 3 个查询服务， 
  r1=geocoderByS1(addr);
  r2=geocoderByS2(addr);
  r3=geocoderByS3(addr);
  // 只要 r1,r2,r3 有一个返回
  // 则返回
  return r1|r2|r3;
}

利用 CompletionService 能够快速实现 Forking 这种集群模式，好比下面的示例代码就展现了具体是如何实现的。首先咱们建立了一个线程池 executor 、一个 CompletionService 对象 cs 和一个Future<Integer>类型的列表 futures，每次经过调用 CompletionService 的 submit() 方法提交一个异步任务，会返回一个 Future 对象，咱们把这些 Future 对象保存在列表 futures 中。经过调用cs.take().get()，咱们可以拿到最快返回的任务执行结果，只要咱们拿到一个正确返回的结果，就能够取消全部任务而且返回最终结果了。

// 建立线程池
ExecutorService executor =
  Executors.newFixedThreadPool(3);
// 建立 CompletionService
CompletionService<Integer> cs =
  new ExecutorCompletionService<>(executor);
// 用于保存 Future 对象
List<Future<Integer>> futures =
  new ArrayList<>(3);
// 提交异步任务，并保存 future 到 futures 
futures.add(
  cs.submit(()->geocoderByS1()));
futures.add(
  cs.submit(()->geocoderByS2()));
futures.add(
  cs.submit(()->geocoderByS3()));
// 获取最快返回的任务执行结果
Integer r = 0;
try {
  // 只要有一个成功返回，则 break
  for (int i = 0; i < 3; ++i) {
    r = cs.take().get();
    // 简单地经过判空来检查是否成功返回
    if (r != null) {
      break;
    }
  }
} finally {
  // 取消全部任务
  for(Future<Integer> f : futures)
    f.cancel(true);
}
// 返回结果
return r;

总结

当须要批量提交异步任务的时候建议你使用 CompletionService。CompletionService 将线程池 Executor 和阻塞队列 BlockingQueue 的功能融合在了一块儿，可以让批量异步任务的管理更简单。除此以外，CompletionService 可以让异步任务的执行结果有序化，先执行完的先进入阻塞队列，利用这个特性，你能够轻松实现后续处理的有序性，避免无谓的等待，同时还能够快速实现诸如 Forking Cluster 这样的需求。