极客时间已完结课程限时免费阅读

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

25 | CompletionService：如何批量执行异步任务？-极客时间

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

2019-04-25 王宝令 来自北京
《Java并发编程实战》
课程介绍

讲述：王宝令

时长06:51大小6.26M

在《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;
欢迎在留言区与我分享你的想法，也欢迎你在留言区记录你的思考过程。感谢阅读，如果你觉得这篇文章对你有帮助的话，也欢迎把它分享给更多的朋友。
分享给需要的人，Ta购买本课程，你将得18元
生成海报并分享

赞 35

提建议

© 版权归极客邦科技所有，未经许可不得传播售卖。 页面已增加防盗追踪，如有侵权极客邦将依法追究其法律责任。
上一篇
24 | CompletableFuture：异步编程没那么难
下一篇
26 | Fork/Join：单机版的MapReduce
unpreview
 写留言

精选留言(85)

  • 张天屹
    2019-04-25
    我觉得问题出在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; }
    展开

    作者回复: 👍

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

    作者回复: 👍

    30
  • 西行寺咕哒子
    2019-04-25
    试过返回值是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(); } 不知道可不可行
    展开

    作者回复: 👍

    共 7 条评论
    30
  • 一道阳光
    2019-04-25
    m.get()和m.set()不是原子性操作，正确代码是:do{int expect = m.get();int min= Integer.min(expect,r);}while(!m.compareAndSet(expect,min))。老师，是这样吗？
    共 1 条评论
    22
  • ipofss
    2019-10-23
    老师，并发工具类，这整个一章，感觉听完似懂非懂的，因为实践中没用过，我要如何弥补这部分，还是说只要听说过，然后用的时候再去查看demo吗

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

    16
  • linqw
    2019-04-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; } }
    展开

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

    7
  • 天涯煮酒
    2019-04-25
    先调用m.get()并跟r比较，再调用m.set()，这里存在竞态条件，线程并不安全
    共 1 条评论
    6
  • 一眼万年
    2019-04-28
    课后思考如果需要等待最小结果，本来就有阻塞队列了，加了个线程池，评论还要加上栏栅，那除了炫技没啥作用
    共 3 条评论
    5
  • 王昊哲
    2019-11-28
    有个疑问：老师也提到那种线程池+阻塞队列实现方式，队列里保存的是任务的结果，而completionService保存的future，那completionService的future拿出来get的时候，也阻塞在get那里了啊，那不跟跟线程池+future的实现一样的弊端了啊？
    共 3 条评论
    4
  • 海鸿
    2019-04-26
    重新发过，刚刚的代码有误！ 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捕获异常后的处理，提高程序鲁棒性
    展开
    4
  • Sunqc
    2019-05-01
    // 获取电商 S1 报价并保存 r=f1.get(); executor.execute(()->save(r)); 如果把r=f1.get（）放进execute里应该是也能保证先执行完的先保存

    作者回复: 是的

    3
  • 黄海峰
    2019-04-25
    我实际测试了第一段代码，确实是异步的，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); }
    展开

    作者回复: 👍

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

    作者回复: 👍，示例代码有问题，已经改了

    3
  • 郑晨Cc
    2019-04-25
    executor.execute（Callable）提交任务是非阻塞的 return m；很大概率返回 Integer.Maxvalue，而且老师为了确保返回这个max还特意加入了save这个阻塞的方法
    4
  • undifined
    2019-04-25
    老师 用 CompletionService 和用 CompletionFuture 查询，然后用 whenComplete 或者 thenAcceptEither 这些方法的区别是什么，我觉得用 CompletionFuture 更直观些； 老师可以在下一讲的时候说一下上一讲的思考题正确答案吗，谢谢老师
    2
  • 空空空空
    2019-04-25
    算低价的时候是用三个不同的线程去计算，是异步的，因此可能算出来并不是预期的结果 老师，这样理解对吗？

    作者回复: 对的！

    2
  • 鸡蛋🎱 达芬奇
    2020-03-20
    扩宽视野，提供思路
    1
  • 梅小西
    2019-10-27
    老师讲的挺不错的，看了这个例子，有几点疑问，还希望老师说明下： // 这个是老师例子： // 创建线程池 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 看来能够做批量处理异步任务的事情，实际应用中，我感觉不太实用！ 以上两点是个人见解，有不对之处请老师指教！
    展开

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

    1
  • ideal sail
    2019-09-11
    课后思考中只要把最小值的比较放在主线程就好了，take一次比较一次，没必要在多个线程里计算最小值。
    1
  • helloworld
    2019-08-30
    老师，冒昧的问下：在文章刚开始的例子，无论是三个询价任务（通过submit方法提交），还是保存询价任务（通过execute方法提交）都是异步的执行执行的啊！如果s1询价的时间过长的话，也不会影响到s2保存保价的先执行啊！他只影响到s1保存询价的动作。老师不知道我说的有么有道理，有问题请老师帮忙指正

    作者回复: 问题出在f1.get()不是异步的，它会阻塞线程

    1