一,深入理解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就是可以将所有任务的返回值返回

总的来说可以分为下面这图所示