一,深入理解CompletableFuture的基本使用
在上一篇文章中讲了线程池,线程任务类是通过实现Runnable实现的,但是Runnable接口会有缺点,一个是不能直接在提交任务之后有返回值,另一个是不能在run方法上面抛异常,因此为了解决这两个问题,jdk中引入了一个新的接口 Callable
1,Callable的基本使用
如直接先定义一个线程任务类Task,实现Callable方法
/**
* @Author: zhenghuisheng
* @Date: 2023/10/19 1:02
*/
public class Task implements Callable {
@Override
public Object call() throws Exception {
return 1;
}
}
随后创建一个Demo类,用于测试。在jdk中,new Thread的参数只能是Runnable类或者其具体的实现类,因此先将Callable类的具体实现作为参数,加入FutureTask中,而FutureTask是一个Runnable的具体的实现类
/**
* @Author: zhenghuisheng
* @Date: 2023/10/19 1:04
*/
public class FutureTaskDemo {
public static void main(String[] args) throws Exception {
//创建一个线程
Task task = new Task();
//将线程作为FutureTask的参数
FutureTask futureTask = new FutureTask(task);
new Thread(futureTask).start();
//将结果返回
System.out.println(futureTask.get());
}
}
这样就成功的将Callable引入进来,成为一个创建线程的一种方式。并且通过这种方式可以将需要的返回值直接获取。
2,Future
上面这张图可以看出这个FutureTask也是Future的一个实现类,接下来查看这个接口中的抽象方法
Future f = new FutureTask(task);
在这个接口中,主要有两个方法,一个是任务执行是否完成,一个是获取任务完成的结果值。get方法在获取到结果之前,内部会进行阻塞
public interface Future<V> {
boolean isDone(); //是否已经执行完成
V get(); //获取执行完的结果
}
如下面这段代码,总共就四个步骤,当线程任务完成之后,会将结果填充到这个FutureTask中,随后通过这个实例获取结果即可。
Task task = new Task(); //构建一个线程任务,实现了callable接口
FutureTask f = new FutureTask(task); //将task类作为参数添加到FutureTask中
threadPool.execute(f); //加入到线程池
System.out.println(f.get()); //获取结果
除了上面最重要的两个方法之外,Future接口中还有下面这些方法
boolean cancel(boolean mayInterruptIfRunning); //取消线程任务
boolean isCancelled(); //判断是否已取消
V get(long timeout, TimeUnit unit) //超时机制获取
3,CompletableFuture
Future可以通过多个异步任务来解决多个同步任务的效率问题,但是其本身也存在着一些缺陷,如无法进行任务与任务之间的链式调用、无法组合多个任务、以及无法在任务处理时做异常处理。为了解决这个问题,因此在juc包中,又引入了一个新的任务类 CompletableFuture
先查看这个CompletableFuture实现类,该类是Future的一个具体实现类,同时还实现了这个 CompletionStage 接口,也就是说该类要全部实现这两个接口中的全部方法,那么该类的方法相必是特别多的,因此该类的功能也非常的强大
public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
}
在这个类中,如果没有自定义线程池,则采用的是 ForkJoinPool 线程池,专门处理cpu密集型任务的线程池。
private static final Executor asyncPool = useCommonPool ?
ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
本文主要讲解的是这个类的使用,因此主要是对内部的一些方法,结合一定的场景来举例。在举例之前,先熟悉一下这些方法中的参数的某些含义。在了解这些规律之后,接下来结合场景对部分api进行讲解
CompletionStage<? extends T> other //表示要创建一个任务
Consumer<? super T> action //表示消费一个任务
Function<? super T,? extends V> fn //表示可以携带上一个任务的返回值到先一个任务
Executor executor //默认的forkjoin或者自定义实现的线程池
3.1,创建CompletableFuture异步操作四种方式
主要分为线程时runnable的实现类和callable的实现类,以及是否自定义线程池等。通过runAsync的方法是没有返回值的,通过supplyAsync的方法是有返回值的,但是在使用get方法时,会进行阻塞。如果没有自定义的实现线程池,则会使用默认的forkjoinpool线程池。
static ThreadPoolExecutor threadPool = ThreadPoolUtil.getThreadPool();
public static void main(String[] args) throws Exception {
//没有返回值,参数为runnable,线程池为forkjoin
CompletableFuture future1 = CompletableFuture.runAsync(() -> System.out.println("run1"));
//没有返回值,参数为runnable,线程池为自定义线程池
CompletableFuture future2 = CompletableFuture.runAsync(() -> System.out.println("run2"),threadPool);
//有返回值,参数为callable,线程池为forkjoin
CompletableFuture future3 = CompletableFuture.supplyAsync(() -> {
return 0;});
//有返回值,参数为callable,线程池为自定义线程池
CompletableFuture future4 = CompletableFuture.supplyAsync(() -> {
return 0;},threadPool);
3.2,get和join获取值
先看这个get方法,如下面这段代码
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
return 0;});
future.get();
由于是线程池,内部肯定要执行对应的run方法,因此定位到这个 AsyncSupply 类,对应的run方法如下。可以发现在执行这个run方法时,会对这个方法进行回调操作
public void run() {
CompletableFuture<T> d; Supplier<T> f;
if ((d = dep) != null && (f = fn) != null) {
dep = null; fn = null;
if (d.result == null) {
try {
d.completeValue(f.get()); //设置值
} catch (Throwable ex) {
d.completeThrowable(ex); //抛异常
}
}
d.postComplete(); //接口回调
}
}
回调的具体实现如下,通过cas的方式对这个 CompletableFuture 类中的result值赋值,随后就可以直接通过get的方式进行获取的操作。
final boolean completeValue(T t) {
//通过cas对result赋值
return UNSAFE.compareAndSwapObject(this, RESULT, null,
(t == null) ? NIL : t);
}
除了get方法能获取到值之外,还能通过join的方式获取值
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
return 0;});
future.join();
join方法的具体实现如下,就是没拿到结果就一直阻塞,拿到才能返回。和get最大的区别就是get使用get方法时,需要手动的抛出异常,而join不需要开发者强制抛出或者捕获异常
public T join() {
Object r;
//拿到结果就返回,没拿到结果就一直阻塞
return reportJoin((r = result) == null ? waitingGet(false) : r);
}
3.3,处理结果whenCompleteAsync
3.3.1,没有异常的情况
当结果获取成功之后,如对某个值的计算,对整体流程都执行成功时,可以使用以下方法,参数同样也是区分了是否需要返回值,是否自定义线程池等
public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwa
ble> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super T
hrowable> action)
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super T
hrowable> action, Executor executor)
举个例子,如当对某个值计算后,成功拿到结果时
public static void main(String[] args) throws Exception {
//创建异步对象
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
int number = 0;
for (int i = 0; i < 100; i++) {
number = number + i;
}
return number;
});
//处理上面的结果
future.whenCompleteAsync(new BiConsumer<Integer,Throwable>() {
@Override
public void accept(Integer data, Throwable throwable) {
System.out.println(data);
}
});
}
3.3.2,有异常时
当结果可能会出现异常时,那么就需要使用到这个 exceptionally 方法
public CompletableFuture<T> exceptionally(Function<Throwable,? extends T>
fn)
接下来对这个方法的使用举例,就是自定义一个简单的异常,随后通过创建的future对象调用获取结果
public static void main(String[] args) throws Exception {
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
int number = 10 / 0;
return number;
});
//处理上面的结果
future.exceptionally(new Function<Throwable, String>() {
@Override
public String apply(Throwable throwable) {
System.out.println("异常信息为:" + throwable.getMessage());
return throwable.getMessage();
}
});
}
没有异常时是需要消费者继续处理消费的,因此参数是一个 BiConsumer 类,而有异常时不需要消费者处理,因此只需创建一个Function处理异常即可。
3.4,多任务链路中的结果处理
3.4.1,thenApply
如果在一个需要多个异步任务的调用链路中,比如B需要A的执行结果,c需要b的执行结果,一直下去,那么就需要使用这个thenApply 了,当然这个方法也区分是否有返回值,是否定义线程池等方法
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
int number = 0;
for (int i = 0; i < 100; i++) {
number = number + i;
}
System.out.println(number);
return number;
}).thenApplyAsync(data -> {
//链路调用1
return data + 999;
}).thenApplyAsync(data -> {
//链路调用2
return data + 888;
});
3.4.2,thenCombine
如果需要结合两个任务的计算,那么可以考虑使用这种thenCombine,比如一个任务算当月的总收入,一个任务算当月的总支出
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
int income = 1000 * 30;
System.out.println("总收入为" + income);
return income;
}).thenCombine(CompletableFuture.supplyAsync(() -> {
int expend = 0;
for (int i = 1; i <= 30; i++) {
expend = expend + i + 500;
}
System.out.println("总支出为:" + expend);
return expend;
}),(income,expend)->{
return income - expend;
});
//获取结果
System.out.println(future.get());
3.4.2,thenAccept
当存在链路调用中,只需关注自身任务的求值,而不需要求总值时,可以直接通过这个thenAccept。如计算一年中走的步数,参数是一个Consumer消费者,会将结果消费,因此在后续的get中,获取到的值为null。
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
int runData = 0;
for (int i = 0; i < 30; i++) {
runData = runData + 10000 + i;
}
System.out.println("第一个月的总步数为:" + runData);
return runData;
}).thenAccept(runData ->{
for (int i = 0; i < 30; i++) {
runData = runData + 10000 + i;
}
System.out.println("前两个月的总步数为:" + runData);
});
System.out.println(future.get());
3.4.3,runAfterEither
如在重试接口中,无论同时发送多少次请求,只要有一个请求成功,就可以不管后续的发出的请求的执行结果
public static void main(String[] args) throws Exception {
CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 1;
});
CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return 2;
});
future1.runAfterEither(future2, new Runnable() {
@Override
public void run() {
System.out.println("已经有一个任务执行完成");
}
}).join();
}
runAfterBoth这个使用和上面的一样,但是得同时满足两个请求
3.4.4,anyOf
原理和上面的一样,就是在多任务中,只要满足一个就可以将对应的请求的返回值返回。而对应的allOf就是可以将所有任务的返回值返回
总的来说可以分为下面这图所示