批量执行异步任务之CompletionService

时间：2020-03-05 23:37:06 阅读：90 评论：0 收藏：0 [点我收藏+]

技术图片

目前一个询价应用，这个应用需要从三个电商询价，然后保存在自己的数据库里。核心示例代码如下，由于是串行的，所以性能很慢

 1 // 向电商 S1 询价，并保存
 2 r1 = getPriceByS1();
 3 save(r1);
 4 
 5 // 向电商 S2 询价，并保存
 6 r2 = getPriceByS2();
 7 save(r2);
 8 
 9 // 向电商 S3 询价，并保存
10 r3 = getPriceByS3();
11 save(r3);

如果采用“ThreadPoolExecutor+Future”的方案，你的优化结果很可能是下面示例代码这样：用三个线程异步执行询价，通过三次调用 Future 的 get() 方法获取询价结果，之后将询价结果保存在数据库中

// 创建线程池
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() 操作上。这点小瑕疵你该如何解决呢？

你应该猜到了，若加一个阻塞队列，获取到各自的报价都进入阻塞队列，然后在主线程中消费阻塞队列，这样就能保证先获取到的报价先保存到数据库了。

// 创建阻塞队列
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));

}

Java SDK 并发包里已经提供了 CompletionService。不但能帮你解决先获取到的报价先保存到数据库的问题，而且还能让代码更简练。其原理也简单：是内部维护了一个阻塞队列，当任务执行结束就把任务的执行结果（Future 对象）加入到阻塞队列中

那么我们来用CompletionService重写刚才的询价系统

// 创建线程池
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));

}