相关阅读
【小家java】java5新特性（简述十大新特性） 重要一跃
【小家java】java6新特性（简述十大新特性） 鸡肋升级
【小家java】java7新特性（简述八大新特性） 不温不火
【小家java】java8新特性（简述十大新特性） 饱受赞誉
【小家java】java9新特性（简述十大新特性） 褒贬不一
【小家java】java10新特性（简述十大新特性） 小步迭代
【小家java】java11新特性（简述八大新特性） 首个重磅LTS版本
前文
开启线程执行任务，不管是使用Runnable（无返回值不支持上报异常）还是Callable（有返回值支持上报异常）接口，都可以轻松实现。那么如果是开启线程池并需要获取结果归集的情况下，如何实现，以及优劣？
本文将分别以这四种方式解决归集的问题，然后看看效率和使用的方便程度即可
1、Futrue
Future接口封装了取消，获取线程结果，以及状态判断是否取消，是否完成这几个方法，都很有用。
Demo：
使用线程池提交Callable接口任务，返回Future接口，添加进list,最后遍历该List且内部使用while轮询,并发获取结果，代码如下
/**
* 使用Futrue来实现多线程执行归集操作
*
* @author fangshixian[email protected]
* @description //
* @date 2018/10/31 11:02
*/
public class FutureDemo {
public static void main(String[] args) {
Long start = Instant.now().toEpochMilli();
//定义一个线程池 方便开启和执行多线程 此处为了方便，直接使用 newFixedThreadPool
ExecutorService exs = Executors.newFixedThreadPool(10);
//结果集 装载在list里面
List<Integer> list = new ArrayList<>();
List<Future<Integer>> futureList = new ArrayList<>();
try {
//1.高速提交10个任务，每个任务返回一个Future入futureList 装载起来 这样10个线程就并行去处理和计算了
for (int i = 0; i < 10; i++) {
futureList.add(exs.submit(new CallableTask(i + 1)));
}
Long getResultStart = Instant.now().toEpochMilli();
System.out.println("结果归集开始时间=" + LocalDateTime.now());
//2.结果归集，用迭代器遍历futureList,高速轮询（模拟实现了并发），任务完成就移除
while (futureList.size() > 0) {
Iterator<Future<Integer>> iterable = futureList.iterator();
//遍历 轮询
while (iterable.hasNext()) {
Future<Integer> future = iterable.next();
//如果任务完成就立马取结果，并且，并且把该任务直接从futureList移除掉 否则判断下一个任务是否完成
if (future.isDone() && !future.isCancelled()) {
//获取结果
Integer i = future.get();
System.out.println("任务i=" + i + "获取完成，移出任务队列！" + LocalDateTime.now());
//把结果装入进去 然后把futrue任务移除
list.add(i);
iterable.remove();
} else {
Thread.sleep(1);//避免CPU高速运转（这就是轮询的弊端），这里休息1毫秒，CPU纳秒级别
}
}
}
System.out.println("list=" + list); //任务的处理结果
System.out.println("总耗时=" + (System.currentTimeMillis() - start) + ",取结果归集耗时=" + (System.currentTimeMillis() - getResultStart));
} catch (Exception e) {
e.printStackTrace();
} finally {
exs.shutdown();
}
}
// 任务 采用sleep模拟处理任务需要消耗的时间
static class CallableTask implements Callable<Integer> {
Integer i; //用来编号任务 方便日志里输出识别
public CallableTask(Integer i) {
super();
this.i = i;
}
@Override
public Integer call() throws Exception {
if (i == 1) {
Thread.sleep(3000);//任务1耗时3秒
} else if (i == 5) {
Thread.sleep(5000);//任务5耗时5秒
} else {
Thread.sleep(1000);//其它任务耗时1秒
}
System.out.println("task线程：" + Thread.currentThread().getName() + "任务i=" + i + ",完成！" + LocalDateTime.now());
return i;
}
}
}
如上图，开启定长为10的线程池：ExecutorService exs = Executors.newFixedThreadPool(10);+任务1耗时3秒，任务5耗时5秒，其他1秒。控制台打印如下：
结果归集开始时间=2018-10-31T11:01:19.457
task线程：pool-1-thread-2任务i=2,完成！2018-10-31T11:01:19.976
task线程：pool-1-thread-4任务i=4,完成！2018-10-31T11:01:19.977
task线程：pool-1-thread-3任务i=3,完成！2018-10-31T11:01:19.977
任务i=4获取完成，移出任务队列！2018-10-31T11:01:19.978
task线程：pool-1-thread-9任务i=9,完成！2018-10-31T11:01:19.978
task线程：pool-1-thread-8任务i=8,完成！2018-10-31T11:01:19.978
task线程：pool-1-thread-7任务i=7,完成！2018-10-31T11:01:19.978
task线程：pool-1-thread-6任务i=6,完成！2018-10-31T11:01:19.978
任务i=6获取完成，移出任务队列！2018-10-31T11:01:19.979
任务i=7获取完成，移出任务队列！2018-10-31T11:01:19.979
task线程：pool-1-thread-10任务i=10,完成！2018-10-31T11:01:19.979
任务i=8获取完成，移出任务队列！2018-10-31T11:01:19.979
任务i=9获取完成，移出任务队列！2018-10-31T11:01:19.979
任务i=10获取完成，移出任务队列！2018-10-31T11:01:19.979
任务i=2获取完成，移出任务队列！2018-10-31T11:01:19.980
任务i=3获取完成，移出任务队列！2018-10-31T11:01:19.980
task线程：pool-1-thread-1任务i=1,完成！2018-10-31T11:01:21.964
任务i=1获取完成，移出任务队列！2018-10-31T11:01:21.965
task线程：pool-1-thread-5任务i=5,完成！2018-10-31T11:01:23.977
任务i=5获取完成，移出任务队列！2018-10-31T11:01:23.979
list=[4, 6, 7, 8, 9, 10, 2, 3, 1, 5]
总耗时=5070,取结果归集耗时=5037
看最后的两个结果输出：
list=[4, 6, 7, 8, 9, 10, 2, 3, 1, 5]--》多执行几遍，最后2个总是1，5最后加进去的，可实现按照任务完成先后顺序获取结果！ 因为1需要3s，5需要5s是最慢的，所以最后进入list
总耗时=5046,取结果归集耗时=5040 ---》符合逻辑，10个任务，定长10线程池，其中一个任务耗时3秒，一个任务耗时5秒，由于并发高速轮训，耗时取最长5秒
建议：此种方法可实现基本目标，任务并行且按照完成顺序获取结果。使用很普遍，老少皆宜，就是CPU有消耗，可以使用！
2、FutureTask
FutureTask是接口RunnableFuture的唯一实现类（实现了Future+Runnable）.
1.Runnable接口，可开启单个线程执行。
2.Future接口，可接受Callable接口的返回值，futureTask.get()阻塞获取结果。
demo：
demo1：两个步骤：1.开启单个线程执行任务，2.阻塞等待执行结果，分离这两步骤，可在这两步中间穿插别的相关业务逻辑
/**
* FutureTask弥补了Future必须用线程池提交返回Future的缺陷，实现功能如下：
* 这两个步骤：一个开启线程执行任务，一个阻塞等待执行结果，分离这两步骤，可在这两步中间穿插别的相关业务逻辑。
*
* @author fangshixian[email protected]
* @description //
* @date 2018/10/31 11:15
*/
public class FutureTaskContorlDemo {
public static void main(String[] args) {
try {
System.out.println("=====例如一个统计公司总部和分部的总利润是否达标100万==========");
//利润 记录总公司的利润综合
Integer count = 0;
//1.定义一个futureTask，假设去远程http获取各个分公司业绩（任务都比较耗时）.
FutureTask<Integer> futureTask = new FutureTask<>(new CallableTask());
Thread futureTaskThread = new Thread(futureTask);
futureTaskThread.start();
System.out.println("futureTaskThread start！" + new Date());
//2.主线程先做点别的事
System.out.println("主线程查询总部公司利润开始时间：" + new Date());
Thread.sleep(5000);
count += 10; //10表示北京集团总部利润。
System.out.println("主线程查询总部公司利润结果时间：" + new Date());
//总部已达标100万利润，就不再继续执行获取分公司业绩任务了
if (count >= 100) {
System.out.println("总部公司利润达标，取消futureTask！" + new Date());
futureTask.cancel(true);//不需要再去获取结果，那么直接取消即可
} else {
System.out.println("总部公司利润未达标，进入阻塞查询分公司利润！" + new Date());
//3总部未达标.阻塞获取，各个分公司结果 然后分别去获取分公司的利润
Integer i = futureTask.get();//真正执行CallableTask
System.out.println("i=" + i + "获取到结果!" + new Date());
}
} catch (Exception e) {
e.printStackTrace();
}
}
// 模拟一个十分耗时的任务 去所有的分公司里去获取利润结果
static class CallableTask implements Callable<Integer> {
@Override
public Integer call() throws Exception {
System.out.println("CallableTask-call，查询分公司利润，执行开始！" + new Date());
Thread.sleep(10000);
System.out.println("CallableTask-call，查询分公司利润，执行完毕！" + new Date());
return 10;
}
}
}
FutureTask这个任务，是Thread.start()的时候就开始执行了的。而结果是.get()的时候才会给你（如果提前完成，结果也会先给你缓存在对象内喽。否则get就会阻塞直到有结果了）
注意：倘若你的任务里抛出了异常。那么get方法就会报错从而
（相当于不需要返回值的异步执行嘛~），但是但是但是，如果你不调用get方法，主线程是不会中断的。中断主线程
输出：
=====例如一个统计公司总部和分部的总利润是否达标100万==========
futureTaskThread start！Wed Oct 31 11:21:33 CST 2018
主线程查询总部公司利润开始时间：Wed Oct 31 11:21:33 CST 2018
CallableTask-call，查询分公司利润，执行开始！Wed Oct 31 11:21:33 CST 2018
主线程查询总部公司利润结果时间：Wed Oct 31 11:21:38 CST 2018
总部公司利润未达标，进入阻塞查询分公司利润！Wed Oct 31 11:21:38 CST 2018
CallableTask-call，查询分公司利润，执行完毕！Wed Oct 31 11:21:43 CST 2018
i=10获取到结果!Wed Oct 31 11:21:43 CST 2018
如上，分离之后，futureTaskThread耗时10秒期间，主线程还穿插的执行了耗时5秒的操作，大大减小总耗时。且可根据业务逻辑实时判断是否需要继续执行futureTask。
Demo2：FutureTask一样可以并发执行任务并获取结果，如下：
/**
* FutureTask实现多线程并发执行任务并取结果归集
*
* @author fangshixian[email protected]
* @description //
* @date 2018/10/31 11:26
*/
public class FutureTaskDemo {
public static void main(String[] args) {
Long start = System.currentTimeMillis();
ExecutorService exs = Executors.newFixedThreadPool(10);
//结果集
List<Integer> list = new ArrayList<>();
List<FutureTask<Integer>> futureList = new ArrayList<>();
try {
//启动线程池 和上面Futrue对比，只有这块有点不一样
for (int i = 0; i < 10; i++) {
FutureTask<Integer> futureTask = new FutureTask<>(new CallableTask(i + 1));
//提交任务，添加返回，Runnable特性
exs.submit(futureTask);
//Future特性 提交任务后 把futureTask添加进futureList
futureList.add(futureTask);
}
Long getResultStart = System.currentTimeMillis();
System.out.println("结果归集开始时间=" + new Date());
//结果归集
while (futureList.size() > 0) {
Iterator<FutureTask<Integer>> iterable = futureList.iterator();
//遍历一遍
while (iterable.hasNext()) {
Future<Integer> future = iterable.next();
if (future.isDone() && !future.isCancelled()) {
//Future特性
Integer i = future.get();
System.out.println("任务i=" + i + "获取完成，移出任务队列！" + new Date());
list.add(i);
//任务完成移除任务
iterable.remove();
} else {
//避免CPU高速轮循，可以休息一下。
Thread.sleep(1);
}
}
}
System.out.println("list=" + list);
System.out.println("总耗时=" + (System.currentTimeMillis() - start) + ",取结果归集耗时=" + (System.currentTimeMillis() - getResultStart));
} catch (Exception e) {
e.printStackTrace();
} finally {
exs.shutdown();
}
}
static class CallableTask implements Callable<Integer> {
Integer i;
public CallableTask(Integer i) {
super();
this.i = i;
}
@Override
public Integer call() throws Exception {
if (i == 1) {
Thread.sleep(3000);//任务1耗时3秒
} else if (i == 5) {
Thread.sleep(5000);//任务5耗时5秒
} else {
Thread.sleep(1000);//其它任务耗时1秒
}
System.out.println("task线程：" + Thread.currentThread().getName() + "任务i=" + i + ",完成！" + new Date());
return i;
}
}
}
输出：
结果归集开始时间=Wed Oct 31 11:26:41 CST 2018
task线程：pool-1-thread-8任务i=8,完成！Wed Oct 31 11:26:42 CST 2018
task线程：pool-1-thread-7任务i=7,完成！Wed Oct 31 11:26:42 CST 2018
task线程：pool-1-thread-6任务i=6,完成！Wed Oct 31 11:26:42 CST 2018
task线程：pool-1-thread-4任务i=4,完成！Wed Oct 31 11:26:42 CST 2018
task线程：pool-1-thread-3任务i=3,完成！Wed Oct 31 11:26:42 CST 2018
task线程：pool-1-thread-2任务i=2,完成！Wed Oct 31 11:26:42 CST 2018
task线程：pool-1-thread-10任务i=10,完成！Wed Oct 31 11:26:42 CST 2018
task线程：pool-1-thread-9任务i=9,完成！Wed Oct 31 11:26:42 CST 2018
任务i=8获取完成，移出任务队列！Wed Oct 31 11:26:42 CST 2018
任务i=9获取完成，移出任务队列！Wed Oct 31 11:26:42 CST 2018
任务i=10获取完成，移出任务队列！Wed Oct 31 11:26:42 CST 2018
任务i=2获取完成，移出任务队列！Wed Oct 31 11:26:42 CST 2018
任务i=3获取完成，移出任务队列！Wed Oct 31 11:26:42 CST 2018
任务i=4获取完成，移出任务队列！Wed Oct 31 11:26:42 CST 2018
任务i=6获取完成，移出任务队列！Wed Oct 31 11:26:42 CST 2018
任务i=7获取完成，移出任务队列！Wed Oct 31 11:26:42 CST 2018
task线程：pool-1-thread-1任务i=1,完成！Wed Oct 31 11:26:44 CST 2018
任务i=1获取完成，移出任务队列！Wed Oct 31 11:26:44 CST 2018
task线程：pool-1-thread-5任务i=5,完成！Wed Oct 31 11:26:46 CST 2018
任务i=5获取完成，移出任务队列！Wed Oct 31 11:26:46 CST 2018
list=[8, 9, 10, 2, 3, 4, 6, 7, 1, 5]
总耗时=5066,取结果归集耗时=5058
建议：
demo1在特定场合例如有十分耗时的业务但有依赖于其他业务不一定非要执行的，可以尝试使用。
demo2多线程并发执行并结果归集，这里多套一层FutureTask比较鸡肋（直接返回Future简单明了）不建议使用。
3、CompletionService
如果你向Executor提交了一个批处理任务，并且希望在它们完成后获得结果,怎么办呢？
Java8之前的做法是让返回
Futrue
，然后调用其get阻塞方法即可。这样做固然可以，但却相当乏味。幸运的是，Java8提供了一个更好的方法：完成服务 (
CompletionService
)。
整合了CompletionService
和Executor
的功能。你可以将BlockingQueue
任务提交给它去执行，然后使用类似于队列中的take和poll方法，在结果完整可用时获得这个结果，像一个使用Callable
打包的BlockingQueue
。Future
CompletionService
是Java8的新增接口，JDK为其提供了一个实现类
ExecutorCompletionService
。
原理:内部通过阻塞队列+FutureTask，实现了任务先完成可优先获取到，即结果按照完成先后顺序排序。
/**
* CompletionService多线程并发任务结果归集
*
* @author fangshixian[email protected]
* @description //
* @date 2018/10/31 11:29
*/
public class CompletionServiceDemo {
public static void main(String[] args) {
Long start = System.currentTimeMillis();
//开启10个线程
ExecutorService exs = Executors.newFixedThreadPool(10);
//结果集
List<Integer> list = new ArrayList<>();
List<Future<Integer>> futureList = new ArrayList<>();
try {
int taskCount = 10;
//1.定义CompletionService ExecutorCompletionService是此接口的唯一实现类 需要把线程池传进去
CompletionService<Integer> completionService = new ExecutorCompletionService<>(exs);
//2.添加任务（向CompletionService添加任务 然后把返回的futrue添加到futureList即可）
for (int i = 0; i < taskCount; i++) {
futureList.add(completionService.submit(new Task(i + 1)));
}
//==================结果归集===================
//方法1：future是提交时返回的，遍历queue则按照任务提交顺序，获取结果 (若是按照提交顺序，那和Futrue的Demo结果将一样，没啥优势可言)
// for (Future<Integer> future : futureList) {
// System.out.println("====================");
// Integer result = future.get();//线程在这里阻塞等待该任务执行完毕,按照
// System.out.println("任务result="+result+"获取到结果!"+new Date());
// list.add(result);
// }
// //方法2.使用内部阻塞队列的take()：内部维护阻塞队列，任务先完成的先获取到
for (int i = 0; i < taskCount; i++) {
Integer result = completionService.take().get();//采用completionService.take()，
System.out.println("任务i==" + result + "完成!" + new Date());
list.add(result);
}
System.out.println("list=" + list);
System.out.println("总耗时=" + (System.currentTimeMillis() - start));
} catch (Exception e) {
e.printStackTrace();
} finally {
exs.shutdown();//关闭线程池
}
}
static class Task implements Callable<Integer> {
Integer i;
public Task(Integer i) {
super();
this.i = i;
}
@Override
public Integer call() throws Exception {
if (i == 5) {
Thread.sleep(5000);
} else {
Thread.sleep(1000);
}
System.out.println("线程：" + Thread.currentThread().getName() + "任务i=" + i + ",执行完成！");
return i;
}
}
}
输出：
线程：pool-1-thread-4任务i=4,执行完成！
任务i==4完成!Wed Oct 31 11:33:35 CST 2018
线程：pool-1-thread-3任务i=3,执行完成！
任务i==3完成!Wed Oct 31 11:33:35 CST 2018
线程：pool-1-thread-2任务i=2,执行完成！
任务i==2完成!Wed Oct 31 11:33:35 CST 2018
线程：pool-1-thread-1任务i=1,执行完成！
任务i==1完成!Wed Oct 31 11:33:35 CST 2018
线程：pool-1-thread-8任务i=8,执行完成！
任务i==8完成!Wed Oct 31 11:33:35 CST 2018
线程：pool-1-thread-7任务i=7,执行完成！
任务i==7完成!Wed Oct 31 11:33:35 CST 2018
线程：pool-1-thread-6任务i=6,执行完成！
任务i==6完成!Wed Oct 31 11:33:35 CST 2018
线程：pool-1-thread-10任务i=10,执行完成！
任务i==10完成!Wed Oct 31 11:33:35 CST 2018
线程：pool-1-thread-9任务i=9,执行完成！
任务i==9完成!Wed Oct 31 11:33:35 CST 2018
线程：pool-1-thread-5任务i=5,执行完成！
任务i==5完成!Wed Oct 31 11:33:39 CST 2018
list=[4, 3, 2, 1, 8, 7, 6, 10, 9, 5] ---》这里证实了确实按照执行完成顺序排序
总耗时=5019 ---》符合逻辑，10个任务，定长5线程池执行，取最长时间。
建议：使用率也挺高，而且能按照完成先后排序，建议如果有排序需求的优先使用。只是多线程并发执行任务结果归集，也可以使用。
优势：能够按照任务完成时间排序，所有有排序需求的，可以考虑使用它。这也是JDK8以前最佳选择
4、CompletableFuture
JDK1.8才新加入的一个实现类，实现了Future, CompletionStage2个接口（CompletionStage接口也是1.8才提供的）
CompletableFuture的简单介绍:：
当一个Future可能需要显示地完成时，使用CompletionStage接口去支持完成时触发的函数和操作。当2个以上线程同时尝试完成、异常完成、取消一个CompletableFuture时，只有一个能成功。
CompletableFuture实现了CompletionStage接口的如下策略：
- 为了完成当前的CompletableFuture接口或者其他完成方法的回调函数的线程，提供了非异步的完成操作
- 没有显式入参Executor的所有async方法都使用ForkJoinPool.commonPool()为了简化监视、调试和跟踪，所有生成的异步任务都是标记接口AsynchronousCompletionTask的实例
- 所有的CompletionStage方法都是独立于其他共有方法实现的，因此一个方法的行为不会受到子类中其他方法的覆盖
CompletableFuture实现了Futurre接口的如下策略：
- CompletableFuture无法直接控制完成，所以cancel操作被视为是另一种异常完成形式。方法isCompletedExceptionally可以用来确定一个CompletableFuture是否以任何异常的方式完成。
- 以一个CompletionException为例，方法get()和get(long,TimeUnit)抛出一个ExecutionException，对应CompletionException。为了在大多数上下文中简化用法，这个类还定义了方法join()和getNow，而不是直接在这些情况中直接抛出CompletionException
CompletionStage接口实现流式编程：
JDK8新增接口，此接口包含38个方法…是的，你没看错，就是38个方法。这些方法主要是为了支持函数式编程中流式处理。
CompletableFuture中4个异步执行任务静态方法：
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
return asyncSupplyStage(asyncPool, supplier);
}
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
Executor executor) {
return asyncSupplyStage(screenExecutor(executor), supplier);
}
public static CompletableFuture<Void> runAsync(Runnable runnable) {
return asyncRunStage(asyncPool, runnable);
}
public static CompletableFuture<Void> runAsync(Runnable runnable,
Executor executor) {
return asyncRunStage(screenExecutor(executor), runnable);
}
如上图，其中supplyAsync用于有返回值的任务，runAsync则用于没有返回值的任务。Executor参数可以手动指定线程池，否则默认ForkJoinPool.commonPool()系统级公共线程池，注意：这些线程都是Daemon线程，主线程结束Daemon线程不结束，只有JVM关闭时，生命周期终止。
组合CompletableFuture：
thenCombine()： 先完成当前CompletionStage和other 2个CompletionStage任务，然后把结果传参给BiFunction进行结果合并操作
三个重载方法如下：
public <U,V> CompletableFuture<V> thenCombine(
CompletionStage<? extends U> other,
BiFunction<? super T,? super U,? extends V> fn) {
return biApplyStage(null, other, fn);
}
public <U,V> CompletableFuture<V> thenCombineAsync(
CompletionStage<? extends U> other,
BiFunction<? super T,? super U,? extends V> fn) {
return biApplyStage(asyncPool, other, fn);
}
public <U,V> CompletableFuture<V> thenCombineAsync(
CompletionStage<? extends U> other,
BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
return biApplyStage(screenExecutor(executor), other, fn);
}
thenCompose()：第一个CompletableFuture执行完毕后，传递给下一个CompletionStage作为入参进行操作。
三个重载方法如下：
public <U> CompletableFuture<U> thenCompose(
Function<? super T, ? extends CompletionStage<U>> fn) {
return uniComposeStage(null, fn);
}
public <U> CompletableFuture<U> thenComposeAsync(
Function<? super T, ? extends CompletionStage<U>> fn) {
return uniComposeStage(asyncPool, fn);
}
public <U> CompletableFuture<U> thenComposeAsync(
Function<? super T, ? extends CompletionStage<U>> fn,
Executor executor) {
return uniComposeStage(screenExecutor(executor), fn);
}
…]… …等等类似的重载方法有很多，后续会专门开博文讲述它的使用方式，请持续关注我的博客
【小家java】Java8新特性之—CompletableFuture的系统讲解和实例演示（使用CompletableFuture构建异步应用）
Demo：
JDK8的CompletableFuture 自带多任务组合方法allOf和anyOf
- allOf是等待所有任务完成，构造后CompletableFuture完成
- anyOf是只要有一个任务完成，构造后CompletableFuture就完成
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
return andTree(cfs, 0, cfs.length - 1);
}
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
return orTree(cfs, 0, cfs.length - 1);
}
方式一：循环创建CompletableFuture list,调用sequence()组装返回一个有返回值的CompletableFuture，返回结果get()获取
/**
* 多线程并发任务,取结果归集
*
* @author fangshixian[email protected]
* @description //
* @date 2018/10/31 11:53
*/
public class CompletableFutureDemo {
public static void main(String[] args) {
Long start = System.currentTimeMillis();
//定长10线程池
ExecutorService exs = Executors.newFixedThreadPool(10);
//结果集
List<String> list = new ArrayList<>();
List<String> list2 = new ArrayList<>();
List<CompletableFuture<String>> futureList = new ArrayList<>();
final List<Integer> taskList = Arrays.asList(2, 1, 3, 4, 5, 6, 7, 8, 9, 10);
try {
//方式一：循环创建CompletableFuture list, 然后组装 组装返回一个有返回值的CompletableFuture，返回结果get()获取
for (int i = 0; i < taskList.size(); i++) {
final int j = i;
//异步执行 拿到每个有返回值的CompletableFuture对象
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> calc(taskList.get(j)), exs)
//Integer转换字符串 thenAccept只接受不返回不影响结果
.thenApply(e -> Integer.toString(e))
//如需获取任务完成先后顺序，此处代码即可
.whenComplete((v, e) -> {
System.out.println("任务" + v + "完成!result=" + v + "，异常 e=" + e + "," + new Date());
list2.add(v);
});
futureList.add(future);
}
//流式获取结果：此处是根据任务添加顺序获取的结果========================
//1.构造一个空CompletableFuture，子任务数为入参任务list size
CompletableFuture<Void> allDoneFuture = CompletableFuture.allOf(futureList.stream()
.filter(f -> f != null).collect(toList()).toArray(new CompletableFuture[futureList.size()]));
//2.流式（总任务完成后，每个子任务join取结果，后转换为list）
list = allDoneFuture.thenApply(v -> futureList.stream().map(CompletableFuture::join).collect(toList())).get();
//流式获取结果：此处是根据任务添加顺序获取的结果========================
System.out.println("任务完成先后顺序，结果list2=" + list2 + "；任务提交顺序，结果list=" + list + ",耗时=" + (System.currentTimeMillis() - start));
} catch (Exception e) {
e.printStackTrace();
} finally {
exs.shutdown();
}
}
//模拟任务的耗时方法
public static Integer calc(Integer i) {
try {
if (i == 1) {
//任务1耗时3秒
Thread.sleep(3000);
} else if (i == 5) {
//任务5耗时5秒
Thread.sleep(5000);
} else {
//其它任务耗时1秒
Thread.sleep(1000);
}
System.out.println("task线程：" + Thread.currentThread().getName() + "任务i=" + i + ",完成！+" + new Date());
} catch (InterruptedException e) {
e.printStackTrace();
}
return i;
}
}
输出：
task线程：pool-1-thread-1任务i=2,完成！+Wed Oct 31 12:27:24 CST 2018
任务2完成!result=2，异常 e=null,Wed Oct 31 12:27:24 CST 2018
task线程：pool-1-thread-4任务i=4,完成！+Wed Oct 31 12:27:24 CST 2018
任务4完成!result=4，异常 e=null,Wed Oct 31 12:27:24 CST 2018
task线程：pool-1-thread-3任务i=3,完成！+Wed Oct 31 12:27:24 CST 2018
任务3完成!result=3，异常 e=null,Wed Oct 31 12:27:24 CST 2018
task线程：pool-1-thread-6任务i=6,完成！+Wed Oct 31 12:27:24 CST 2018
任务6完成!result=6，异常 e=null,Wed Oct 31 12:27:24 CST 2018
task线程：pool-1-thread-8任务i=8,完成！+Wed Oct 31 12:27:24 CST 2018
任务8完成!result=8，异常 e=null,Wed Oct 31 12:27:24 CST 2018
task线程：pool-1-thread-7任务i=7,完成！+Wed Oct 31 12:27:24 CST 2018
任务7完成!result=7，异常 e=null,Wed Oct 31 12:27:24 CST 2018
task线程：pool-1-thread-10任务i=10,完成！+Wed Oct 31 12:27:24 CST 2018
任务10完成!result=10，异常 e=null,Wed Oct 31 12:27:24 CST 2018
task线程：pool-1-thread-9任务i=9,完成！+Wed Oct 31 12:27:24 CST 2018
任务9完成!result=9，异常 e=null,Wed Oct 31 12:27:24 CST 2018
task线程：pool-1-thread-2任务i=1,完成！+Wed Oct 31 12:27:26 CST 2018
任务1完成!result=1，异常 e=null,Wed Oct 31 12:27:26 CST 2018
task线程：pool-1-thread-5任务i=5,完成！+Wed Oct 31 12:27:28 CST 2018
任务5完成!result=5，异常 e=null,Wed Oct 31 12:27:28 CST 2018
任务完成先后顺序，结果list2=[2, 4, 3, 6, 8, 7, 10, 9, 1, 5]；任务提交顺序，结果list=[2, 1, 3, 4, 5, 6, 7, 8, 9, 10],耗时=5141
方式二：全流式处理转换成CompletableFuture[]+allOf组装成一个无返回值CompletableFuture，join等待执行完毕。返回结果whenComplete获取。—》推荐
/**
* 多线程并发任务,取结果归集
*
* @author fangshixian[email protected]
* @description //
* @date 2018/10/31 11:53
*/
public class CompletableFutureDemo {
public static void main(String[] args) {
Long start = System.currentTimeMillis();
//定长10线程池
ExecutorService exs = Executors.newFixedThreadPool(10);
//结果集
List<String> list = new ArrayList<>();
List<String> list2 = new ArrayList<>();
final List<Integer> taskList = Arrays.asList(2, 1, 3, 4, 5, 6, 7, 8, 9, 10);
try {
//方式二：全流式处理转换成CompletableFuture[]+组装成一个无返回值CompletableFuture，join等待执行完毕。返回结果whenComplete获取
CompletableFuture<Integer>[] cfs = taskList.stream().map(i ->
//把计算任务 交给CompletableFuture异步去处理执行
CompletableFuture.supplyAsync(() -> calc(i), exs)
// 把计算完成结果做Function处理：此处是转换成了字符串
.thenApply(h -> Integer.toString(h))
//如需获取任务完成先后顺序，此处代码即可 会先处理先完成的任务 后处理后完成的任务 使用起来比CompletionService确实方便不少
.whenComplete((v, e) -> {
System.out.println("任务" + v + "完成!result=" + v + "，异常 e=" + e + "," + new Date());
list2.add(v);
})).toArray(CompletableFuture[]::new); //此处直接toArray 不toList了
//等待总任务完成，但是封装后无返回值，必须自己whenComplete()获取 此处使用join来获取结果
CompletableFuture.allOf(cfs).join();
System.out.println("任务完成先后顺序，结果list2=" + list2 + "；任务提交顺序，结果list=" + list + ",耗时=" + (System.currentTimeMillis() - start));
} catch (Exception e) {
e.printStackTrace();
} finally {
exs.shutdown();
}
}
//模拟任务的耗时方法
public static Integer calc(Integer i) {
try {
if (i == 1) {
//任务1耗时3秒
Thread.sleep(3000);
} else if (i == 5) {
//任务5耗时5秒
Thread.sleep(5000);
} else {
//其它任务耗时1秒
Thread.sleep(1000);
}
System.out.println("task线程：" + Thread.currentThread().getName() + "任务i=" + i + ",完成！+" + new Date());
} catch (InterruptedException e) {
e.printStackTrace();
}
return i;
}
}
输出：
task线程：pool-1-thread-7任务i=7,完成！+Wed Oct 31 12:19:41 CST 2018
task线程：pool-1-thread-3任务i=3,完成！+Wed Oct 31 12:19:41 CST 2018
task线程：pool-1-thread-4任务i=4,完成！+Wed Oct 31 12:19:41 CST 2018
任务4完成!result=4，异常 e=null,Wed Oct 31 12:19:41 CST 2018
task线程：pool-1-thread-8任务i=8,完成！+Wed Oct 31 12:19:41 CST 2018
任务8完成!result=8，异常 e=null,Wed Oct 31 12:19:41 CST 2018
task线程：pool-1-thread-9任务i=9,完成！+Wed Oct 31 12:19:41 CST 2018
task线程：pool-1-thread-1任务i=2,完成！+Wed Oct 31 12:19:41 CST 2018
task线程：pool-1-thread-10任务i=10,完成！+Wed Oct 31 12:19:41 CST 2018
任务2完成!result=2，异常 e=null,Wed Oct 31 12:19:41 CST 2018
任务10完成!result=10，异常 e=null,Wed Oct 31 12:19:41 CST 2018
任务3完成!result=3，异常 e=null,Wed Oct 31 12:19:41 CST 2018
任务7完成!result=7，异常 e=null,Wed Oct 31 12:19:41 CST 2018
task线程：pool-1-thread-6任务i=6,完成！+Wed Oct 31 12:19:41 CST 2018
任务9完成!result=9，异常 e=null,Wed Oct 31 12:19:41 CST 2018
任务6完成!result=6，异常 e=null,Wed Oct 31 12:19:41 CST 2018
task线程：pool-1-thread-2任务i=1,完成！+Wed Oct 31 12:19:43 CST 2018
任务1完成!result=1，异常 e=null,Wed Oct 31 12:19:43 CST 2018
task线程：pool-1-thread-5任务i=5,完成！+Wed Oct 31 12:19:45 CST 2018
任务5完成!result=5，异常 e=null,Wed Oct 31 12:19:45 CST 2018
任务完成先后顺序，结果list2=[4, 8, 2, 10, 3, 7, 9, 6, 1, 5]；任务提交顺序，结果list=[],耗时=5166 ---》符合逻辑，10个任务，10个线程并发执行，其中任务1耗时3秒，任务5耗时5秒，耗时取最大值。
建议：CompletableFuture满足并发执行，顺序完成先手顺序获取的目标。而且支持每个任务的异常返回，配合流式编程，用起来速度飞起。JDK源生支持，API丰富，推荐使用。
总结
本文从原理、demo、建议三个方向分析了常用多线程并发，取结果归集的几种实现方案，希望对大家有所启发，整理表格如下：
关注A哥
| Author | A哥(YourBatman) |
|---|---|
| 个人站点 | www.yourbatman.cn |
| [email protected] | |
| 微 信 | fsx641385712 |
|  |
| 公众号 | BAT的乌托邦（ID：BAT-utopia） |
| 知识星球 | BAT的乌托邦 |
| 每日文章推荐 | 每日文章推荐 |