您当前的位置: 首页 > 

教校花学妹JDK批量异步任务最强工具CompletionService

发布时间:2021-04-18 23:49:03 ,浏览量:0

如何优化一个查询各个价格接口的代码?若使用“ThreadPoolExecutor+Future”,可能优化如下:

三个线程异步执行查询价格,通过三次调用Future的get()方法获取结果,之后将查询结果保存在MySQL。

若获取price1耗时很长,那么即便获取price2耗时短,也无法让保存price2的操作先执行,因为主线程都阻塞在f1.get()。这种问题如何解决呢?

加个阻塞队列! 获取到price1、2、3都进入阻塞队列,然后在主线程消费阻塞队列,就能保证先获取到的价格先保存:

CompletionService实现查询价格

实际开发推荐CompletionService,不但能帮你解决先获取到的价格先保存,还能精简代码。

CompletionService内部维护了一个阻塞队列,当任务执行结束就把任务的执行结果入队,但CompletionService是把任务执行结果的Future对象入队,而上面demo是把任务最终执行结果入队。

创建CompletionService

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

  • ExecutorCompletionService(Executor executor);
  • ExecutorCompletionService(Executor executor, BlockingQueuecompletionQueue)

这俩构造器都需要传入一个线程池,若不指定completionQueue,默认使用无界LinkedBlockingQueue。任务执行结果的Future对象就是加入到completionQueue中。

学长~你直接给我写段代码解释清楚点呗

让我们试着利用CompletionService实现高性能的查询房价系统。 之后通过CompletionService#submit()提交三个询价操作,这三个询价操作将会被CompletionService异步执行。

最后CompletionService#take()获取一个Future对象(加入到阻塞队列的是任务执行结果的Future对象),调用Future#get()就能返回执行结果。

CompletionService接口

CompletionService接口提供的方法 submit()相关的方法有两个:

  • 一个方法参数是Callabletask
  • 一个方法有两个参数,分别是Runnable task和V result,该方法类似于ThreadPoolExecutor的Futuresubmit(Runnable task, T result),
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类型的列表 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和阻塞队列融合,让批量异步任务管理更简单。

CompletionService能让异步任务的执行结果有序化,先执行完的先进入阻塞队列,利用该特性,可以轻松实现后续处理的有序性,避免无谓等待,同时还可以快速实现诸如Forking Cluster这样的需求。

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

关注
打赏
1688896170
查看更多评论

暂无认证

  • 0浏览

    0关注

    115984博文

    0收益

  • 0浏览

    0点赞

    0打赏

    0留言

私信
关注
热门博文
立即登录/注册

微信扫码登录

0.0964s