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

当需要批量提交异步任务的时候建议使用CompletionService。 CompletionService 将线程池 Executor 和阻塞队列 BlockingQueue 的功能融合在了一起，能够让批量异步任务的管理更简单。 除此之外，CompletionService 能够让异步任务的执行结果有序化，先执行完的先进入阻塞队列，利用这个特性可以轻松实现后续处理的有序性，避免无谓的等待，同时还可以快速实现诸如 Forking Cluster 这样的需求。 CompletionService 的实现类 ExecutorCompletionService，需要自己创建线程池，可以让多个 ExecutorCompletionService 的线程池隔离，这种隔离性能避免几个特别耗时的任务拖垮整个应用的风险。

利用 CompletionService 实现询价系统

CompletionService 的实现原理是内部维护了一个阻塞队列，当任务执行结束就把任务的执行结果加入到阻塞队列中，CompletionService 是把任务执行结果的 Future 对象加入到阻塞队列中。 CompletionService 接口的实现类是 ExecutorCompletionService，这个实现类的构造方法有两个，分别是： ExecutorCompletionService(Executor executor)； ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)。 这两个构造方法都需要传入一个线程池，如果不指定 completionQueue，那么默认会使用无界的 LinkedBlockingQueue。 任务执行结果的 Future 对象就是加入到 completionQueue 中。 下面的示例代码完整地展示了如何利用 CompletionService 来实现高性能的询价系统。 通过 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 接口提供的方法有 5 个，这 5 个方法的方法签名如下所示。 其中，submit() 相关的方法有两个。 一个方法参数是 Callable<V> task。 另外一个方法有两个参数，分别是 Runnable task 和 V result，这个方法类似于 ThreadPoolExecutor 的 <T> Future<T> submit(Runnable task, T result) CompletionService 接口其余的 3 个方法，都是和阻塞队列相关的，take()、poll() 都是从阻塞队列中获取并移除一个元素；它们的区别在于如果阻塞队列是空的，那么调用 take() 方法的线程会被阻塞，而 poll() 方法会返回 null 值。 poll(long timeout, TimeUnit unit) 方法支持以超时的方式获取并移除阻塞队列头部的一个元素，如果等待了 timeout unit 时间，阻塞队列还是空的，那么该方法会返回 null 值。