25 | CompletionService：如何批量执行异步任务？

在《23 | Future：如何用多线程实现最优的“烧水泡茶”程序？》的最后，我给你留了道思考题，如何优化一个询价应用的核心代码？如果采用“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() 操作上。这点小瑕疵你该如何解决呢？

估计你已经想到了，增加一个阻塞队列，获取到S1、S2、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，这个实现类的构造方法有两个，分别是：

ExecutorCompletionService(Executor executor)；
ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)。

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

下面的示例代码完整地展示了如何利用CompletionService来实现高性能的询价系统。其中，我们没有指定completionQueue，因此默认使用无界的LinkedBlockingQueue。之后通过CompletionService接口提供的submit()方法提交了三个询价操作，这三个询价操作将会被CompletionService异步执行。最后，我们通过CompletionService接口提供的take()方法获取一个Future对象（前面我们提到过，加入到阻塞队列中的是任务执行结果的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个方法的方法签名如下所示。

其中，submit()相关的方法有两个。一个方法参数是Callable<V> task，前面利用CompletionService实现询价系统的示例代码中，我们提交任务就是用的它。另外一个方法有两个参数，分别是Runnable task和V result，这个方法类似于ThreadPoolExecutor的 <T> Future<T> submit(Runnable task, T result) ，这个方法在《23 | Future：如何用多线程实现最优的“烧水泡茶”程序？》中我们已详细介绍过，这里不再赘述。

CompletionService接口其余的3个方法，都是和阻塞队列相关的，take()、poll()都是从阻塞队列中获取并移除一个元素；它们的区别在于如果阻塞队列是空的，那么调用 take() 方法的线程会被阻塞，而 poll() 方法会返回 null 值。 poll(long timeout, TimeUnit unit) 方法支持以超时的方式获取并移除阻塞队列头部的一个元素，如果等待了 timeout unit时间，阻塞队列还是空的，那么该方法会返回 null 值。

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实现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这样的需求。

CompletionService的实现类ExecutorCompletionService，需要你自己创建线程池，虽看上去有些啰嗦，但好处是你可以让多个ExecutorCompletionService的线程池隔离，这种隔离性能避免几个特别耗时的任务拖垮整个应用的风险。

课后思考

本章使用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());
// 将询价结果异步保存到数据库
// 并计算最低报价
AtomicReference<Integer> m =
  new AtomicReference<>(Integer.MAX_VALUE);
for (int i=0; i<3; i++) {
  executor.execute(()->{
    Integer r = null;
    try {
      r = cs.take().get();
    } catch (Exception e) {}
    save(r);
    m.set(Integer.min(m.get(), r));
  });
}
return m;

欢迎在留言区与我分享你的想法，也欢迎你在留言区记录你的思考过程。感谢阅读，如果你觉得这篇文章对你有帮助的话，也欢迎把它分享给更多的朋友。

精选留言

张天屹

2019-04-25 09:27:51

我觉得问题出在return m这里需要等待三个线程执行完成，但是并没有。
...
AtomicReference<Integer> m = new AtomicReference<>(Integer.MAX_VALUE);
CountDownLatch latch = new CountDownLatch(3);
for(int i=0; i<3; i++) {
executor.execute(()->{
Integer r = null;
try {
r = cs.take().get();
} catch(Exception e) {}
save(r);
m.set(Integer.min(m.get(), r));
latch.countDown();
});
latch.await();
return m;
}

作者回复

👍

2019-04-26 20:24:54
小华

2019-04-25 14:14:11

看老师的意图是要等三个比较报假的线程都执行完才能执行主线程的的return m，但是代码无法保证三个线程都执行完，和主线程执行return的顺序，因此，m的值不是准确的，可以加个线程栈栏，线程执行完计数器，来达到这效果

作者回复

👍

2019-04-25 19:24:58
西行寺咕哒子

2019-04-25 10:24:16

试过返回值是2147483647，也就是int的最大值。没有等待操作完成就猴急的返回了。 m.set(Integer.min(m.get(), r)... 这个操作也不是原子操作。
试着自己弄了一下：
public Integer run(){
// 创建线程池
ExecutorService executor = Executors.newFixedThreadPool(3);
// 创建 CompletionService
CompletionService<Integer> cs = new ExecutorCompletionService<>(executor);
AtomicReference<Integer> m = new AtomicReference<>(Integer.MAX_VALUE);
// 异步向电商 S1 询价
cs.submit(()->getPriceByS1());
// 异步向电商 S2 询价
cs.submit(()->getPriceByS2());
// 异步向电商 S3 询价
cs.submit(()->getPriceByS3());
// 将询价结果异步保存到数据库
// 并计算最低报价
for (int i=0; i<3; i++) {
Integer r = logIfError(()->cs.take().get());
executor.execute(()-> save(r));
m.getAndUpdate(v->Integer.min(v, r));
}
return m.get();
}
不知道可不可行

作者回复

👍

2019-04-26 21:12:55
一道阳光

2019-04-25 09:58:15

m.get()和m.set()不是原子性操作，正确代码是:do{int expect = m.get();int min= Integer.min(expect,r);}while(!m.compareAndSet(expect,min))。老师，是这样吗？
ipofss

2019-10-23 11:02:16

老师，并发工具类，这整个一章，感觉听完似懂非懂的，因为实践中没用过，我要如何弥补这部分，还是说只要听说过，然后用的时候再去查看demo吗

作者回复

用的时候查文档就行，工具类主要是会用，知道什么场景用什么

2019-10-23 11:53:45
linqw

2019-04-25 20:57:25

老师stampedLock的获取锁源码，老师能帮忙解惑下么？阻塞的读线程cowait是挂在写节点的下方么？老师能解惑下基于的理论模型
private long acquireWrite(boolean interruptible, long deadline) {
WNode node = null, p;
for (int spins = -1;;) { // spin while enqueuing
long m, s, ns;
//如果当前的state是无锁状态即100000000
if ((m = (s = state) & ABITS) == 0L) {
//设置成写锁
if (U.compareAndSwapLong(this, STATE, s, ns = s + WBIT))
return ns;
}
else if (spins < 0)
//当前锁状态为写锁状态，并且队列为空，设置自旋值
spins = (m == WBIT && wtail == whead) ? SPINS : 0;
else if (spins > 0) {
//自旋操作，就是让线程在此自旋
if (LockSupport.nextSecondarySeed() >= 0)
--spins;
}
//如果队列尾元素为空，初始化队列
else if ((p = wtail) == null) { // initialize queue
WNode hd = new WNode(WMODE, null);
if (U.compareAndSwapObject(this, WHEAD, null, hd))
wtail = hd;
}
//当前要加入的元素为空，初始化当前元素，前置节点为尾节点
else if (node == null)
node = new WNode(WMODE, p);
//队列的稳定性判断，当前的前置节点是否改变，重新设置
else if (node.prev != p)
node.prev = p;
//将当前节点加入尾节点中
else if (U.compareAndSwapObject(this, WTAIL, p, node)) {
p.next = node;
break;
}
}

作者回复

这可难倒我了，并发库的源码我只是零散得看的，看完基本也忘得差不多了，感觉自己也不是搞算法的料，放弃了😂

2019-04-26 20:23:01
天涯煮酒

2019-04-25 09:05:02

先调用m.get()并跟r比较，再调用m.set()，这里存在竞态条件，线程并不安全
王昊哲

2019-11-28 13:20:11

有个疑问：老师也提到那种线程池+阻塞队列实现方式，队列里保存的是任务的结果，而completionService保存的future，那completionService的future拿出来get的时候，也阻塞在get那里了啊，那不跟跟线程池+future的实现一样的弊端了啊？
一眼万年

2019-04-28 09:48:42

课后思考如果需要等待最小结果，本来就有阻塞队列了，加了个线程池，评论还要加上栏栅，那除了炫技没啥作用
海鸿

2019-04-26 19:05:08

重新发过，刚刚的代码有误！
1.for循环线程池执行属于异步导致未等比价结果就 return了，需要等待三次比价结果才能 return，可以用 CountDownLatch
2. m. set( Integer. min( m. get(), r))存在竞态条件，可以更改为
Integer o;
do{
o= m. get();
if(o<=r){ break;}
}
while(! m. compareAndSet( o, r));
3.还有一个小问题就是 try- catch捕获异常后的处理，提高程序鲁棒性
郑晨Cc

2019-04-25 02:46:53

executor.execute（Callable）提交任务是非阻塞的 return m；很大概率返回 Integer.Maxvalue，而且老师为了确保返回这个max还特意加入了save这个阻塞的方法
Sunqc

2019-05-01 08:30:43

// 获取电商 S1 报价并保存
r=f1.get();
executor.execute(()->save(r));

如果把r=f1.get（）放进execute里应该是也能保证先执行完的先保存

作者回复

是的

2019-05-01 16:42:57
黄海峰

2019-04-25 10:08:08

我实际测试了第一段代码，确实是异步的，f1.get不会阻塞主线程。。。

public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(3);
Future<Integer> f1 = executor.submit(()->getPriceByS1());
Future<Integer> f2 = executor.submit(()->getPriceByS2());
Future<Integer> f3 = executor.submit(()->getPriceByS3());

executor.execute(()-> {
try {
save(f1.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
});
executor.execute(()-> {
try {
save(f2.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
});
executor.execute(()-> {
try {
save(f3.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
});
}

private static Integer getPriceByS1() {
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
}
private static Integer getPriceByS2() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 2;
}
private static Integer getPriceByS3() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 3;
}
private static void save(Integer i) {
System.out.println("save " + i);
}

作者回复

👍

2019-04-25 19:14:15
Corner

2019-04-25 09:12:31

1.AtomicReference<Integer>的get方法应该改成使用cas方法
2.最后筛选最小结果的任务是异步执行的，应该在return之前做同步，所以最好使用sumit提交该任务便于判断任务的完成
最后请教老师一下，第一个例子中为什么主线程会阻塞在f1.get()方法呢？

作者回复

👍，示例代码有问题，已经改了

2019-04-25 20:09:54
undifined

2019-04-25 09:05:02

老师用 CompletionService 和用 CompletionFuture 查询，然后用 whenComplete 或者 thenAcceptEither 这些方法的区别是什么，我觉得用 CompletionFuture 更直观些；
老师可以在下一讲的时候说一下上一讲的思考题正确答案吗，谢谢老师
空空空空

2019-04-25 08:55:50

算低价的时候是用三个不同的线程去计算，是异步的，因此可能算出来并不是预期的结果
老师，这样理解对吗？

作者回复

对的！

2019-04-28 23:20:22
litblue

2023-02-22 09:54:50

以下是我尝试使用ChatGPT来回答的这个问题。
回答如下：

该代码创建一个包含 3 个线程的线程池，以异步的方式向三个电商网站 S1、S2 和 S3 发起询价请求，并使用 CompletionService 将询价结果收集起来。然后，使用 executor 来将结果异步保存到数据库，并计算最低报价。
在计算最低报价时，使用了 AtomicReference 来保存当前最低报价。每个线程都会将自己获得的报价保存到数据库中，并更新最低报价。
该代码的设计上遵循了并发编程中的一些最佳实践，如使用线程池来管理线程、使用 CompletionService 来收集异步任务的结果、使用 AtomicReference 来保证线程安全等。同时，该代码使用 Lambda 表达式来简化了代码结构。
需要注意的是，该代码中对异常的处理不够严谨。如果在询价或保存时出现异常，该异常将会被忽略，无法得到及时的处理。因此，需要在实际应用中根据具体情况添加必要的异常处理机制。
另外一个关键问题：代码中的返回值问题。
当前代码中的返回值是在主线程中计算的最低报价，这样做可能存在一定的问题。因为线程池中的任务是异步执行的，所以在主线程中返回最低报价时，可能有一些任务还没有完成，因此可能会得到一个不完整的结果。
解决这个问题的方法是使用 Future 来异步获取线程池中每个任务的计算结果，然后在所有任务都执行完毕后，再返回最终的最低报价。
以下是一个修改过的示例代码:

ExecutorService executor = Executors.newFixedThreadPool(3);
List<Future<Integer>> futures = new ArrayList<>();

futures.add(executor.submit(() -> getPriceByS1()));
futures.add(executor.submit(() -> getPriceByS2()));
futures.add(executor.submit(() -> getPriceByS3()));

AtomicInteger minPrice = new AtomicInteger(Integer.MAX_VALUE);
for (Future<Integer> future : futures) {
try {
int price = future.get();
save(price);
minPrice.set(Integer.min(minPrice.get(), price));
} catch (InterruptedException | ExecutionException e) {
// 处理异常
}
}
executor.shutdown();

return minPrice.get();
鸡蛋🎱 达芬奇

2020-03-20 20:18:32

扩宽视野，提供思路
梅小西

2019-10-27 23:05:00

老师讲的挺不错的，看了这个例子，有几点疑问，还希望老师说明下：
// 这个是老师例子：

// 创建线程池
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应该是要绑定泛型，代表异步任务的返回结果，实际应用中，几乎不太可能所有的异步任务的返回类型是一样的，除非设置成Object这种通用型，那又会导致拿到结果后需要强转，代码看起来更难受；
其次，对于返回的结果的处理方式，实际应用中几乎也是不同的，那就要针对每一个take出来的结果做判断，这实际上也是会导致代码很难维护；

综上，CompletionService 看来能够做批量处理异步任务的事情，实际应用中，我感觉不太实用！

以上两点是个人见解，有不对之处请老师指教！

作者回复

每种异步任务都会创建一个新的，不可能所有的异步任务都用一个。如果某些需要共用，可以包装一个类就可以了

2019-10-29 08:42:14
ideal sail

2019-09-11 16:51:32

课后思考中只要把最小值的比较放在主线程就好了，take一次比较一次，没必要在多个线程里计算最小值。