CompletableFuture 简介
CompletableFuture 在 Java 里面被用于异步编程,异步通常意味着非阻塞,可以使得我们的任务单独运行在与主线程分离的其他线程中,并且通过回调可以在主线程中得到异步任务的执行状态,是否完成,和是否异常等信息。
CompletableFuture 实现了 Future、CompletionStage 接口
实现了 Future 接口就可以兼容现在有线程池框架,而 CompletionStage 接口才是异步编程的接口抽象,里面定义多种异步方法,通过这两者集合,从而打造出了强大的 CompletableFuture 类。
Future 与 CompletableFuture
Futrue 在 Java 里面,通常用来表示一个异步任务的引用,比如我们将任务提交到线程池里面,然后我们会得到一个 Futrue,在 Future 里面有 isDone 方法来判断任务是否处理结束,还有 get 方法可以一直阻塞直到任务结束然后获取结果,但整体来说这种方式,还是同步的,因为需要客户端不断阻塞等待或者不断轮询才能知道任务是否完成。
Future 的主要缺点如下:
(1)不支持手动完成
我提交了一个任务,但是执行太慢了,我通过其他路径已经获取到了任务结果,现在没法把这个任务结果通知到正在执行的线程,所以必须主动取消或者一直等待它执行完成
(2)不支持进一步的非阻塞调用
通过 Future 的 get 方法会一直阻塞到任务完成,但是想在获取任务之后执行额外的任务,因为 Future 不支持回调函数,所以无法实现这个功能
(3)不支持链式调用
对于 Future 的执行结果,我们想继续传到下一个 Future 处理使用,从而形成一个链式的 pipline 调用,这在 Future 中是没法实现的。
(4)不支持多个 Future 合并
比如我们有 10 个 Future 并行执行,我们想在所有的 Future 运行完毕之后,执行某些函数,是没法通过 Future 实现的。
(5)不支持异常处理
Future 的 API 没有任何的异常处理的 api,所以在异步运行时,如果出了问题是不好定位的。
CompletableFuture 入门
使用 CompletableFuture
场景:主线程里面创建一个 CompletableFuture,然后主线程调用future.get()
方法获取子线程执行结果会阻塞,最后我们在一个子线程中使其终止。
用途:已经知道子线程的执行结果了,那么就不需要等待子线程执行完成了,直接结束它。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26
|
public static void main(String[] args) throws Exception { CompletableFuture<String> future = new CompletableFuture<>(); new Thread(() -> { try { System.out.println(Thread.currentThread().getName() + "开始工作"); Thread.sleep(3000); System.out.println("子线程工作结束,取消阻塞"); future.complete("success"); } catch (Exception e) { e.printStackTrace(); } }, "子线程").start(); System.out.println(Thread.currentThread().getName() + "线程阻塞"); System.out.println("主线程调用get方法获取结果为: " + future.get()); System.out.println("主线程完成,阻塞结束!!!!!!"); }
|
没有返回值的异步任务 runAsync
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
|
public static void main(String[] args) throws Exception { System.out.println("主线程开始"); CompletableFuture<Void> future = CompletableFuture.runAsync(() -> { try { System.out.println("子线程启动干活"); Thread.sleep(3000); System.out.println("子线程完成"); } catch (Exception e) { e.printStackTrace(); } }); System.out.println("主线程取得结果:"+future.get()); System.out.println("主线程结束"); }
|
有返回值的异步任务 supplyAsync
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
|
public static void main(String[] args) throws Exception { System.out.println("主线程开始"); CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { try { System.out.println("子线程开始任务"); Thread.sleep(3000); } catch (Exception e) { e.printStackTrace(); } return "子线程完成了!"; }); System.out.println("主线程取得结果:"+future.get()); System.out.println("主线程结束"); }
|
线程依赖 thenApply
当一个线程依赖另一个线程时,可以使用 thenApply()
方法来把这两个线程串行化。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27
| private static Integer num = 10;
public static void main(String[] args) throws Exception { System.out.println("主线程开始"); CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { try { System.out.println(Thread.currentThread().getName() + "开始执行加10任务"); num += 10; } catch (Exception e) { e.printStackTrace(); } return num;
}).thenApply(integer -> { System.out.println(Thread.currentThread().getName() + "开始执行平方任务"); return num * num; });
Integer integer = future.get(); System.out.println("主线程结束, 子线程的结果为:" + integer); }
|
结果
还有一个相似的方法thenApplyAsync()
1 2 3 4 5 6 7 8 9 10 11 12 13
| CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { try { System.out.println(Thread.currentThread().getName() + "开始执行加10任务"); num += 10; } catch (Exception e) { e.printStackTrace(); } return num;
}).thenApplyAsync(integer -> { System.out.println(Thread.currentThread().getName() + "开始执行平方任务"); return num * num; });
|
结果
两者的区别
- thenApplyAsync:使用同一个线程执行第一个以及后面依赖的所有的任务;
- thenApply:由调用的程序去执行任务;
消费处理结果 thenAccept
thenAccept 消费处理结果,接收任务的处理结果,并消费处理,无返回结果。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
| private static Integer num = 10;
public static void main(String[] args) { System.out.println("主线程开始"); CompletableFuture.supplyAsync(() -> { try { System.out.println("加10任务开始"); num += 10; } catch (Exception e) { e.printStackTrace(); } return num; }).thenApply(integer -> num * num) .thenAccept(integer -> System.out.println("子线程全部处理完成,最后调用了accept,结果为:" + integer)); }
|
thenRun()方法
thenRun: 一般也用于回调函数最后的执行,但这个方法不接受回调函数的返回值,纯粹就代表执行任务的最后一个步骤
1 2 3 4 5 6 7 8 9 10 11 12 13 14
| private static Integer num = 10;
public static void main(String[] args) { System.out.println("主线程开始"); CompletableFuture.supplyAsync(() -> { try { System.out.println("加10任务开始"); num += 10; } catch (Exception e) { e.printStackTrace(); } return num; }).thenApply(integer -> num * num).thenRun(() -> System.out.println("任务的最后一个步骤")); }
|
异常处理 exceptionally
exceptionally 异常处理,出现异常时触发
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
| private static Integer num = 10;
public static void main(String[] args) throws Exception { System.out.println("主线程开始"); CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { int i = 1 / 0; System.out.println("加10任务开始"); num += 10; return num; }).exceptionally(ex -> { System.out.println(ex.getMessage()); return -1; }); System.out.println(future.get()); }
|
handle 类似于thenAccept/thenRun
方法,是最后一步的处理调用,但是同时可以处理异常
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21
| private static Integer num = 10;
public static void main(String[] args) throws Exception{ System.out.println("主线程开始"); CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { System.out.println("加10任务开始"); num += 10; return num; }).handle((i,ex) ->{ System.out.println("进入handle方法");
if(ex != null){ System.out.println("发生了异常,内容为:" + ex.getMessage()); return -1; }else{ System.out.println("正常完成,内容为: " + i); return i; } }); System.out.println(future.get()); }
|
结果合并 thenCompose
thenCompose 合并两个有依赖关系的 CompletableFutures 的执行结果
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
| private static Integer num = 10;
public static void main(String[] args) throws Exception{ System.out.println("主线程开始"); CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { System.out.println("加10任务开始"); num += 10; return num; }); CompletableFuture<Integer> future1 = future.thenCompose(i -> CompletableFuture.supplyAsync(() -> i + 1)); System.out.println(future.get()); System.out.println(future1.get()); }
|
结果合并 thenCombine
两个没有依赖关系的 CompletableFutures 任务
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29
| private static Integer num = 10;
public static void main(String[] args) throws Exception { System.out.println("主线程开始");
CompletableFuture<Integer> job1 = CompletableFuture.supplyAsync(() -> { System.out.println("加10任务开始"); num += 10; return num; });
CompletableFuture<Integer> job2 = CompletableFuture.supplyAsync(() -> { System.out.println("乘10任务开始"); num = num * 10; return num; });
CompletableFuture<Object> future = job1.thenCombine(job2, new BiFunction<Integer, Integer, List<Integer>>() { @Override public List<Integer> apply(Integer a, Integer b) { List<Integer> list = new ArrayList<>(); list.add(a); list.add(b); return list; } }); System.out.println("合并结果为:" + future.get()); }
|
同步多个任务
allOf:一系列独立的 future 任务,等其所有的任务执行完后做一些事情
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39
| private static Integer num = 10;
public static void main(String[] args) throws Exception{ System.out.println("主线程开始"); List<CompletableFuture> list = new ArrayList<>(); CompletableFuture<Integer> job1 = CompletableFuture.supplyAsync(() -> { System.out.println("加10任务开始"); num += 10; return num; }); list.add(job1);
CompletableFuture<Integer> job2 = CompletableFuture.supplyAsync(() -> { System.out.println("乘以10任务开始"); num = num * 10; return num; }); list.add(job2);
CompletableFuture<Integer> job3 = CompletableFuture.supplyAsync(() -> { System.out.println("减以10任务开始"); num = num * 10; return num; }); list.add(job3);
CompletableFuture<Integer> job4 = CompletableFuture.supplyAsync(() -> { System.out.println("除以10任务开始"); num = num * 10; return num; }); list.add(job4); List<Integer> collect = list.stream().map(CompletableFuture<Integer>::join).collect(Collectors.toList()); System.out.println(collect); }
|
执行结果
anyOf: 只要在多个 future 里面有一个返回,整个任务就可以结束,而不需要等到每一个 future 结束
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60
| private static Integer num = 10;
public static void main(String[] args) throws Exception{ System.out.println("主线程开始"); CompletableFuture<Integer>[] futures = new CompletableFuture[4]; CompletableFuture<Integer> job1 = CompletableFuture.supplyAsync(() -> { try{ Thread.sleep(5000); System.out.println("加10任务开始"); num += 10; return num; }catch (Exception e){ return 0; } }); futures[0] = job1;
CompletableFuture<Integer> job2 = CompletableFuture.supplyAsync(() -> { try{ Thread.sleep(2000); System.out.println("乘以10任务开始"); num = num * 10; return num; }catch (Exception e){ return 1; }
}); futures[1] = job2;
CompletableFuture<Integer> job3 = CompletableFuture.supplyAsync(() -> { try{ Thread.sleep(3000); System.out.println("减以10任务开始"); num = num * 10; return num; }catch (Exception e){ return 2; }
}); futures[2] = job3;
CompletableFuture<Integer> job4 = CompletableFuture.supplyAsync(() -> { try{ Thread.sleep(4000); System.out.println("除以10任务开始"); num = num * 10; return num; }catch (Exception e){ return 3; } }); futures[3] = job4; CompletableFuture<Object> future = CompletableFuture.anyOf(futures); System.out.println(future.get()); }
|
执行结果