JDK1.8以后,Concurrent包提供了一个异步编程工具CompletableFuture,用它可以取代JDK1.8之前的Future+callable形式
简单用法 complete(…)函数CompletableFuturefuture = new CompletableFuture<>(); new Thread(() ->{ try { System.out.println("线程开始执行 "+new Date()); Thread.sleep(2000); System.out.println("获取到我要的结果 "+new Date()); future.complete("我要的结果"); Thread.sleep(3000); System.out.println("线程执行结束 "+new Date()); } catch (InterruptedException e) { e.printStackTrace(); } }).start(); String s = future.get(); System.out.println(s+" "+new Date());
线程A中调用future.complete(…),线程B中future.get()即可获取A中传入的参数。以此来完成线程间的通信(即B获取A的执行结果)。
只要A中不调用future.complete(…),那么B中的future.get()会一直阻塞
不需要执行结果的runAsync(…)函数CompletableFuturefuture = CompletableFuture.runAsync(() -> { try { System.out.println("子线程开始执行"+new Date()); Thread.sleep(2000); System.out.println("子线程执行结束"+new Date()); } catch (InterruptedException e) { e.printStackTrace(); } }); System.out.println("主线程开始get"+new Date()); future.get(); System.out.println("主线程get结束"+new Date());
提交一个Runnable异步任务,future.get()会阻塞到任务执行结束。
需要执行结果的supplyAsync(…)CompletableFuturefuture = CompletableFuture.supplyAsync(new Supplier () { @SneakyThrows @Override public String get() { System.out.println(new Date()+"任务开始执行"); Thread.sleep(2000); System.out.println(new Date()+"任务执行结束"); return "带返回结果"; } }); System.out.println(new Date()+"主线程获取返回值"); String s = future.get(); System.out.println(new Date()+"主线程获取返回值="+s); //lambda表达式写法 CompletableFuture future1 = CompletableFuture.supplyAsync(() -> "返回数据");
这个方法就可以替代Callable和Future的组合。Supplier对像类似于callable对象,需要实现其get方法并返回结果,结果为泛型参数,可以为任何对象。
然后同过future.get()即可获得supplier对象返回的结果
相比于Future+Callable的组合,CompletableFuture提供了三个额外的回调函数。
sleep任务,下面demo中会用到
private void sleep(String name){
try {
System.out.println(new Date() + name+"开始执行" + Thread.currentThread().getName());
Thread.sleep(3000);
System.out.println(new Date() + name+"执行结束");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
不依赖任务执行参数,不需要返回结果,只再任务结束后的后续处理thenRun()函数
CompletableFuturefuture = CompletableFuture.supplyAsync((Supplier ) () -> { sleep("任务"); return "返回数据"; }).thenRun(() ->{ sleep("thenAccept"); }); System.out.println(new Date()+"主线程开始等待"+Thread.currentThread().getName()); future.get(); System.out.println(new Date()+"主线程等待结束");
任务结束后会执行thenRun函数,直到thenRun函数执行结束前,future.get()都会处于阻塞状态
依赖任务执行参数,不需要返回结果的回调thenAccept(…)函数CompletableFuturefuture = CompletableFuture.supplyAsync((Supplier ) () -> { sleep("任务"); return "返回数据"; }).thenAccept((param) -> { System.out.println("thenAccept获取到参数"+param); sleep("thenAccept"); }); System.out.println(new Date()+"主线程开始等待"+Thread.currentThread().getName()); future.get(); System.out.println(new Date()+"主线程等待结束");
thenAccept参数是一个函数接口Consumer,接口参数为任务的返回信息。
依赖任务执行参数,需要返回结果的回调thenApply(…)函数如果任务没有返回结果,param参数为null(Void)
CompletableFuturefuture = CompletableFuture.supplyAsync((Supplier ) () -> { sleep("任务"); return "返回数据"; }).thenApply((Function ) param -> { System.out.println("thenApply获取到参数"+param); sleep("thenApply"); return param.length(); }); System.out.println(new Date()+"主线程开始等待"+Thread.currentThread().getName()); Integer integer = future.get(); System.out.println(new Date()+"主线程等待结束"+integer);
thenApply也是个函数接口Function
,P代表任务返回的参数类型,param是任务返回的参数,R为回调返回的参数类型。demo中回调返回的参数类型为Integer类型
题外话 ForkJoinPoolCompletableFuture采用ForkJoinPool实现。
ForkJoinPool就是JDK7提供的一种“分治算法”的多线程并行计算框架。Fork意为分叉,Join意为合
并,一分一合,相互配合,形成分治算法。此外,也可以将ForkJoinPool看作一个单机版的
Map/Reduce,多个线程并行计算。
相比于ThreadPoolExecutor,ForkJoinPool可以更好地实现计算的负载均衡,提高资源利用率。
利用ForkJoinPool,可以把大的任务拆分成很多小任务,然后这些小任务被所有的线程执行,从而
实现任务计算的负载均衡。



