Future接口定义了操作 异步任务执行一些方法 ，如获取异步任务的执行结果、取消任务的执行、判断任务是否被取消、判断任务执行是否完毕等。
Callable接口中定义了需要有返回的任务需要实现的方法。
比如主线程让一个子线程去执行任务，子线程可能比较耗时，启动子线程开始执行任务后，
主线程就去做其他事情了，过了一会才去获取子任务的执行结果。
2、从之前的FutureTask说开去 2.1、本源的Future接口相关架构 2.2、Code-1package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
FutureTask futureTask = new FutureTask<>(() -> {
System.out.println("-----come in FutureTask");
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "" + ThreadLocalRandom.current().nextInt(100);
});
Thread t1 = new Thread(futureTask, "t1");
t1.start();
//3 秒钟后才出来结果，还没有计算你提前来拿 ( 只要一调用 get 方法，对于结果就是不见不散，会导致阻塞 )
// System.out.println(Thread.currentThread().getName() + "t" + futureTask.get());
// 3 秒钟后才出来结果，我只想等待 1 秒钟，过时不候
System.out.println(Thread.currentThread().getName() + "t" + futureTask.get(1L, TimeUnit.SECONDS));
System.out.println(Thread.currentThread().getName() + " t " + " run... here");
}
}
get()阻塞
一旦调用get()方法，不管是否计算完成都会导致阻塞，o(╥﹏╥)o
2.3、Code-2
package com.atguigu.juc.cf;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class FutureTaskDemo
{
public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException
{
FutureTask futureTask = new FutureTask<>(() -> {
System.out.println(Thread.currentThread().getName() + "t" + "---come in");
try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); }
return 1024;
});
new Thread(futureTask,"t1").start();
//System.out.println(futureTask.get());//不见不散，只要出现get方法，不管是否计算完成都阻塞等待结果出来再运行
//System.out.println(futureTask.get(2L,TimeUnit.SECONDS));//过时不候
//不要阻塞，尽量用轮询替代
while(true)
{
if(futureTask.isDone())
{
System.out.println("----result: "+futureTask.get());
break;
}else{
System.out.println("还在计算中，别催，越催越慢，再催熄火");
}
}
}
}
isDone()轮询
轮询的方式会耗费无谓的CPU资源，而且也不见得能及时地得到计算结果.
如果想要异步获取结果,通常都会以轮询的方式去获取结果
尽量不要阻塞
2.4、想完成一些复杂的任务
① 应对Future的完成时间，完成了可以告诉我，也就是我们的回调通知
② 将两个异步计算合成一个异步计算，这两个异步计算互相独立，同时第二个又依赖第一个的结果。
③ 当Future集合中某个任务最快结束时，返回结果。
④ 等待Future结合中的所有任务都完成。
⑤ 。。。。。。
3、对Future的改进 3.1、CompletableFuture和CompletionStage源码分别介绍 3.1.1、类架构说明 3.1.2、接口CompletionStage- CompletionStage代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段
- 一个阶段的计算执行可以是一个Function,Consumer或者Runnable。比如：stage.thenApply(x->(x -System.out.print(x)).thenRun(O->System.out.println))
- 一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发
- 代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段，有些类似Linux系统的管道分隔符传参数。
- 在Java8中，CompletableFuture提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合CompletableFuture的方法。
- 它可能代表一个明确完成的Future,也有可能代表一个完成阶段(CompletionStage),它支持在计算完成以后触发一些函数或执行某些动作。
- 它实现了Future和CompletionStage接口
public static CompletableFuture runAsync(Runnable runnable)
public static CompletableFuture runAsync(Runnable runnable,Executor executor)
3.2.2、supplyAsync 有 返回值public static CompletableFuture supplyAsync(Supplier supplier)
public static CompletableFuture supplyAsync(Supplier supplier,Executor executor)
3.2.3、上述Executor executor参数说明没有指定Executor的方法，直接使用默认的ForkJoinPool.commonPool() 作为它的线程池执行异步代码。
如果指定线程池，则使用我们自定义的或者特别指定的线程池执行异步代码
3.2.4、Code无返回值
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture future = CompletableFuture.runAsync(() -> {
System.out.println(Thread.currentThread().getName() + " t " + "-----come in");
// 暂停几秒钟线程
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("-----task is over");
});
System.out.println(future.get());
}
}
有 返回值
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " t " + "-----come in");
// 暂停几秒钟线程
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return ThreadLocalRandom.current().nextInt(100);
});
System.out.println(completableFuture.get());
}
}
3.2.5、Code之通用演示，减少阻塞和轮询
从Java8开始引入了CompletableFuture，它是Future的功能增强版，
可以传入回调对象，当异步任务完成或者发生异常时，自动调用回调对象的回调方法
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " t " + "-----come in");
int result = ThreadLocalRandom.current().nextInt(10);
// 暂停几秒钟线程
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("----- 计算结束耗时 1 秒钟， result ： " + result);
if (result > 6) {
int age = 10 / 0;
}
return result;
}).whenComplete((v, e) -> {
if (e == null) {
System.out.println("-----result: " + v);
}
}).exceptionally(e -> {
System.out.println("-----exception: " + e.getCause() + " t " + e.getMessage());
return -44;
});
// 主线程不要立刻结束，否则 CompletableFuture 默认使用的线程池会立刻关闭 : 暂停 3 秒钟线程
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
3.2.6、CompletableFuture的优点
- 异步任务结束时，会自动回调某个对象的方法；
- 异步任务出错时，会自动回调某个对象的方法；
- 主线程设置好回调后，不再关心异步任务的执行，异步任务之间可以顺序执行
Lambda +Stream+链式调用+Java8函数式编程带走
4.2、先说说join和get对比
join=get, 一样的，区别就是 join 不抛出异常
4.3、业务需求说明对内微服务多系统调用
对外网站比价
经常出现在等待某条 SQL 执行完成后，再继续执行下一条 SQL ，而这两条 SQL 本身是并无关系的，可以同时进行执行的。
我们希望能够两条 SQL 同时进行处理，而不是等待其中的某一条 SQL 完成后，再继续下一条。同理，
对于分布式微服务的调用，按照实际业务，如果是无关联step by step的业务，可以尝试是否可以多箭齐发，同时调用。
我们去比同一个商品在各个平台上的价格，要求获得一个清单列表， 1 step by step，查完京东查淘宝，查完淘宝查天猫......
2 all 一口气同时查询。。。。。
一波流Java8函数式编程带走
package com.atguigu.juc.cf;
import lombok.Getter;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
public class CompletableFutureNetMallDemo
{
static List list = Arrays.asList(
new NetMall("jd"),
new NetMall("pdd"),
new NetMall("taobao"),
new NetMall("dangdangwang"),
new NetMall("tmall")
);
//同步 ，step by step
public static List getPriceByStep(List list,String productName)
{
return list
.stream().
map(netMall -> String.format(productName + " in %s price is %.2f", netMall.getMallName(), netMall.calcPrice(productName)))
.collect(Collectors.toList());
}
//异步 ，多箭齐发
public static List getPriceByASync(List list,String productName)
{
return list
.stream()
.map(netMall -> CompletableFuture.supplyAsync(() -> String.format(productName + " is %s price is %.2f", netMall.getMallName(), netMall.calcPrice(productName))))
.collect(Collectors.toList())
.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList());
}
public static void main(String[] args)
{
long startTime = System.currentTimeMillis();
List list1 = getPriceByStep(list, "mysql");
for (String element : list1) {
System.out.println(element);
}
long endTime = System.currentTimeMillis();
System.out.println("----costTime: "+(endTime - startTime) +" 毫秒");
System.out.println();
long startTime2 = System.currentTimeMillis();
List list2 = getPriceByASync(list, "mysql");
for (String element : list2) {
System.out.println(element);
}
long endTime2 = System.currentTimeMillis();
System.out.println("----costTime: "+(endTime2 - startTime2) +" 毫秒");
}
}
class NetMall
{
@Getter
private String mallName;
public NetMall(String mallName)
{
this.mallName = mallName;
}
public double calcPrice(String productName)
{
//检索需要1秒钟
try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
return ThreadLocalRandom.current().nextDouble() * 2 + productName.charAt(0);
}
}
4.4、CompletableFuture常用方法
4.4.1、获得结果和触发计算
获取结果 不见不散 public T get() 过时不候 public T get(long timeout, TimeUnit unit)
没有计算完成的情况下，给我一个替代结果
立即获取结果不阻塞 ,计算完，返回计算完成后的结果,没算完，返回设定的valueIfAbsent值
public T getNow(T valueIfAbsent)
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 533;
});
// 去掉注释上面计算没有完成，返回 444
// 开启注释上满计算完成，返回计算结果
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(completableFuture.getNow(444));
}
}
public T join()
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println(CompletableFuture.supplyAsync(() -> "abc").thenApply(r -> r + "123").join());
}
}
4.4.2、对计算结果进行处理
thenApply
计算结果存在依赖关系，这两个线程串行化
由于存在依赖关系(当前步错，不走下一步)，当前步骤有异常的话就叫停。
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 当一个线程依赖另一个线程时用 thenApply 方法来把这两个线程串行化 ,
CompletableFuture.supplyAsync(() -> {
// 暂停几秒钟线程
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("111");
return 1024;
}).thenApply(f -> {
System.out.println("222");
return f + 1;
}).thenApply(f -> {
int age = 10 / 0;
// 异常情况：那步出错就停在那步。
System.out.println("333");
return f + 1;
}).whenCompleteAsync((v, e) -> {
System.out.println("*****v: " + v);
}).exceptionally(e -> {
e.printStackTrace();
return null;
});
System.out.println("----- 主线程结束， END");
// 主线程不要立刻结束，否则 CompletableFuture 默认使用的线程池会立刻关闭 :
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
handle
有异常也可以往下一步走，根据带的异常参数可以进一步处理
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException { // 当一个线程依赖另一个线程时用 handle 方法来把这两个线程串行化 ,
// 异常情况：有异常也可以往下一步走，根据带的异常参数可以进一步处理
CompletableFuture.supplyAsync(() -> {
// 暂停几秒钟线程
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("111");
return 1024;
}).handle((f, e) -> {
int age = 10 / 0;
System.out.println("222");
return f + 1;
}).handle((f, e) -> {
System.out.println("333");
return f + 1;
}).whenCompleteAsync((v, e) -> {
System.out.println("*****v: " + v);
}).exceptionally(e -> {
e.printStackTrace();
return null;
});
System.out.println("----- 主线程结束， END");
// 主线程不要立刻结束，否则 CompletableFuture 默认使用的线程池会立刻关闭 :
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
总结：
whenComplete和whenCompleteAsync的区别：
whenComplete:是执行当前任务的线程执行继续执行whenComplete的任务。
whenCompleteAsync:是执行把whenCompleteAsync这个任务继续提交给线程池来进行执行。
4.4.3、对计算结果进行消费接收任务的处理结果，并消费处理，无返回结果
thenAccept
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture.supplyAsync(() -> {
return 1;
}).thenApply(f -> {
return f + 2;
}).thenApply(f -> {
return f + 3;
}).thenApply(f -> {
return f + 4;
}).thenAccept(r -> System.out.println(r));
}
}
补充：Code之任务之间的顺序执行
① thenRun：thenRun(Runnable runnable) 任务 A 执行完执行 B，并且 B 不需要 A 的结果
② thenAccept：thenAccept(Consumer action) 任务 A 执行完执行 B，B 需要 A 的结果，但是任务 B 无返回值
③ thenApply：thenApply(Function fn) 任务 A 执行完执行 B，B 需要 A 的结果，同时任务 B 有返回值
System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenRun(() -> {
}).join());
System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenAccept(resultA -> {
}).join());
System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenApply(resultA -> resultA + " resultB").join());
返回结果：
null
null
resultA resultB
4.4.4、对计算速度进行选用
谁快用谁
applyToEither
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture completableFuture1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in ");
// 暂停几秒钟线程
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 10;
});
CompletableFuture completableFuture2 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in ");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 20;
});
CompletableFuture thenCombineResult = completableFuture1.applyToEither(completableFuture2, f -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in ");
return f + 1;
});
System.out.println(Thread.currentThread().getName() + " t " + thenCombineResult.get());
}
}
返回结果
ForkJoinPool.commonPool-worker-1 ---come in
ForkJoinPool.commonPool-worker-2 ---come in
ForkJoinPool.commonPool-worker-2 ---come in
main 21
4.4.5、对计算结果进行合并
两个CompletionStage任务都完成后，最终能把两个任务的结果一起交给thenCombine 来处理
先完成的先等着，等待其它分支任务 thenCombine
code标准版，好理解先拆分
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture completableFuture1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in ");
return 10;
});
CompletableFuture completableFuture2 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in ");
return 20;
});
CompletableFuture thenCombineResult = completableFuture1.thenCombine(completableFuture2, (x, y) -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in ");
return x + y;
});
System.out.println(thenCombineResult.get());
}
}
返回结果：
ForkJoinPool.commonPool-worker-1 ---come in
ForkJoinPool.commonPool-worker-1 ---come in
main ---come in
30
package com.atguigu.juc.cf;
import java.util.concurrent.*;
public class CompletableFutureDemo2 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture thenCombineResult = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in 1");
return 10;
}).thenCombine(CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in 2");
return 20;
}), (x, y) -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in 3");
return x + y;
}).thenCombine(CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in 4");
return 30;
}), (a, b) -> {
System.out.println(Thread.currentThread().getName() + " t " + "---come in 5");
return a + b;
});
System.out.println("----- 主线程结束， END");
System.out.println(thenCombineResult.get());
// 主线程不要立刻结束，否则 CompletableFuture 默认使用的线程池会立刻关闭 :
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
返回结果：
ForkJoinPool.commonPool-worker-1 ---come in 1
ForkJoinPool.commonPool-worker-2 ---come in 2
main ---come in 3
ForkJoinPool.commonPool-worker-2 ---come in 4
main ---come in 5
----- 主线程结束， END
60
