Java 8 的异步编程利器 CompletableFuture 详解

Posted ThinkWon

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Java 8 的异步编程利器 CompletableFuture 详解相关的知识,希望对你有一定的参考价值。

文章目录

最近刚好使用CompeletableFuture优化了项目中的代码,所以跟大家一起学习CompletableFuture

一个例子回顾 Future

因为CompletableFuture实现了Future接口,我们先来回顾Future吧

Future是Java5新加的一个接口,它提供了一种异步并行计算的功能。如果主线程需要执行一个很耗时的计算任务,我们就可以通过future把这个任务放到异步线程中执行。主线程继续处理其他任务,异步线程处理完成后,再通过Future获取计算结果

来看个简单例子吧,假设我们有两个任务服务,一个查询用户基本信息,一个是查询用户勋章信息。如下

public class UserInfoService 

    public UserInfo getUserInfo(Long userId) throws InterruptedException 
        Thread.sleep(300);//模拟调用耗时
        return new UserInfo("666", "技术人成长之路", 27); //一般是查数据库,或者远程调用返回的
    


public class MedalService 
    public MedalInfo getMedalInfo(long userId) throws InterruptedException 
        Thread.sleep(500); //模拟调用耗时
        return new MedalInfo("666", "守护勋章");
    

接下来,我们来演示下,在主线程中是如何使用Future来进行异步调用的。

public class FutureTest 

    public static void main(String[] args) throws ExecutionException, InterruptedException 
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        UserInfoService userInfoService = new UserInfoService();
        MedalService medalService = new MedalService();
        long userId =666L;
        long startTime = System.currentTimeMillis();

        //调用用户服务获取用户基本信息
        FutureTask<UserInfo> userInfoFutureTask = new FutureTask<>(new Callable<UserInfo>() 
            @Override
            public UserInfo call() throws Exception 
                return userInfoService.getUserInfo(userId);
            
        );
        executorService.submit(userInfoFutureTask);

        Thread.sleep(300); //模拟主线程其它操作耗时

        FutureTask<MedalInfo> medalInfoFutureTask = new FutureTask<>(new Callable<MedalInfo>() 
            @Override
            public MedalInfo call() throws Exception 
                return medalService.getMedalInfo(userId);
            
        );
        executorService.submit(medalInfoFutureTask);

        UserInfo userInfo = userInfoFutureTask.get();//获取个人信息结果
        MedalInfo medalInfo = medalInfoFutureTask.get();//获取勋章信息结果

        System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
    

运行结果:

总共用时806ms

如果我们不使用Future进行并行异步调用,而是在主线程串行进行的话,耗时大约为300+500+300 = 1100 ms。可以发现,future+线程池异步配合,提高了程序的执行效率。

但是Future对于结果的获取,不是很友好,只能通过阻塞或者轮询的方式得到任务的结果。

  • Future.get() 就是阻塞调用,在线程获取结果之前get方法会一直阻塞
  • Future提供了一个isDone方法,可以在程序中轮询这个方法查询执行结果。

阻塞的方式和异步编程的设计理念相违背,而轮询的方式会耗费无谓的CPU资源。因此,JDK8设计出CompletableFuture。CompletableFuture提供了一种观察者模式类似的机制,可以让任务执行完成后通知监听的一方。

一个例子走进CompletableFuture

我们还是基于以上Future的例子,改用CompletableFuture 来实现

public class FutureTest 

    public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException 
        UserInfoService userInfoService = new UserInfoService();
        MedalService medalService = new MedalService();
        long userId =666L;
        long startTime = System.currentTimeMillis();

        //调用用户服务获取用户基本信息
        CompletableFuture<UserInfo> completableUserInfoFuture = CompletableFuture.supplyAsync(() -> userInfoService.getUserInfo(userId));

        Thread.sleep(300); //模拟主线程其它操作耗时

        CompletableFuture<MedalInfo> completableMedalInfoFuture = CompletableFuture.supplyAsync(() -> medalService.getMedalInfo(userId)); 

        UserInfo userInfo = completableUserInfoFuture.get(2,TimeUnit.SECONDS);//获取个人信息结果
        MedalInfo medalInfo = completableMedalInfoFuture.get();//获取勋章信息结果
        System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
    

可以发现,使用CompletableFuture,代码简洁了很多。CompletableFuture的supplyAsync方法,提供了异步执行的功能,线程池也不用单独创建了。实际上,它CompletableFuture使用了默认线程池是ForkJoinPool.commonPool

CompletableFuture提供了几十种方法,辅助我们的异步任务场景。这些方法包括创建异步任务、任务异步回调、多个任务组合处理等方面。我们一起来学习吧

CompletableFuture使用场景

创建异步任务

CompletableFuture创建异步任务,一般有supplyAsync和runAsync两个方法

创建异步任务

  • supplyAsync执行CompletableFuture任务,支持返回值
  • runAsync执行CompletableFuture任务,没有返回值。

supplyAsync方法

//使用默认内置线程池ForkJoinPool.commonPool(),根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
//自定义线程,根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

runAsync方法

//使用默认内置线程池ForkJoinPool.commonPool(),根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable) 
//自定义线程,根据runnable构建执行任务
public static CompletableFuture<Void> runAsync(Runnable runnable,  Executor executor)

实例代码如下:

public class FutureTest 

    public static void main(String[] args) 
        //可以自定义线程池
        ExecutorService executor = Executors.newCachedThreadPool();
        //runAsync的使用
        CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> System.out.println("run,技术人成长之路"), executor);
        //supplyAsync的使用
        CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> 
                    System.out.print("supply,技术人成长之路");
                    return "技术人成长之路"; , executor);
        //runAsync的future没有返回值,输出null
        System.out.println(runFuture.join());
        //supplyAsync的future,有返回值
        System.out.println(supplyFuture.join());
        executor.shutdown(); // 线程池需要关闭
    


//输出
run,技术人成长之路
null
supply,技术人成长之路技术人成长之路

任务异步回调

1. thenRun/thenRunAsync

public CompletableFuture<Void> thenRun(Runnable action);
public CompletableFuture<Void> thenRunAsync(Runnable action);

CompletableFuture的thenRun方法,通俗点讲就是,做完第一个任务后,再做第二个任务。某个任务执行完成后,执行回调方法;但是前后两个任务没有参数传递,第二个任务也没有返回值

public class FutureThenRunTest 

    public static void main(String[] args) throws ExecutionException, InterruptedException 
        CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
                ()->
                    System.out.println("先执行第一个CompletableFuture方法任务");
                    return "技术人成长之路";
                
        );

        CompletableFuture thenRunFuture = orgFuture.thenRun(() -> 
            System.out.println("接着执行第二个任务");
        );

        System.out.println(thenRunFuture.get());
    


//输出
先执行第一个CompletableFuture方法任务
接着执行第二个任务
null

thenRun 和 thenRunAsync 有什么区别呢?可以看下源码哈:

private static final Executor asyncPool = useCommonPool ?
    ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

public CompletableFuture<Void> thenRun(Runnable action) 
    return uniRunStage(null, action);


public CompletableFuture<Void> thenRunAsync(Runnable action) 
    return uniRunStage(asyncPool, action);

如果你执行第一个任务的时候,传入了一个自定义线程池:

  • 调用thenRun方法执行第二个任务时,则第二个任务和第一个任务是共用同一个线程池
  • 调用thenRunAsync方法执行第二个任务时,则第一个任务使用的是你自己传入的线程池,第二个任务使用的是ForkJoin线程池

TIPS:后面介绍的thenAccept和thenAcceptAsync,thenApply和thenApplyAsync等,它们之间的区别也是这个哈。

2.thenAccept/thenAcceptAsync

CompletableFuture的thenAccept方法表示,第一个任务执行完成后,执行第二个回调方法任务,会将第一个任务的执行结果,作为第二个任务的入参,传递到回调方法中,但是回调方法是没有返回值的。

public class FutureThenAcceptTest 

    public static void main(String[] args) throws ExecutionException, InterruptedException 
        CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
                ()->
                    System.out.println("原始CompletableFuture方法任务");
                    return "技术人成长之路";
                
        );

        CompletableFuture thenAcceptFuture = orgFuture.thenAccept((a) -> 
            if ("技术人成长之路".equals(a)) 
                System.out.println("关注了");
            

            System.out.println("先考虑考虑");
        );

        System.out.println(thenAcceptFuture.get());
    

3. thenApply/thenApplyAsync

CompletableFuture的thenApply方法表示,第一个任务执行完成后,执行第二个回调方法任务,会将第一个任务的执行结果,作为第二个任务的入参,传递到回调方法中,并且回调方法是有返回值的。

public class FutureThenApplyTest 

    public static void main(String[] args) throws ExecutionException, InterruptedException 
        CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
                ()->
                    System.out.println("原始CompletableFuture方法任务");
                    return "技术人成长之路";
                
        );

        CompletableFuture<String> thenApplyFuture = orgFuture.thenApply((a) -> 
            if ("技术人成长之路".equals(a)) 
                return "关注了";
            

            return "先考虑考虑";
        );

        System.out.println(thenApplyFuture.get());
    


//输出
原始CompletableFuture方法任务
关注了

4. exceptionally

CompletableFuture的exceptionally方法表示,某个任务执行异常时,执行的回调方法;并且有抛出异常作为参数,传递到回调方法。

public class FutureExceptionTest 

    public static void main(String[] args) throws ExecutionException, InterruptedException 
        CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
                ()->
                    System.out.println("当前线程名称:" + Thread.currentThread().getName());
                    throw new RuntimeException();
                
        );

        CompletableFuture<String> exceptionFuture = orgFuture.exceptionally((e) -> 
            e.printStackTrace();
            return "你的程序异常啦";
        );

        System.out.println(exceptionFuture.get());
    


//输出
当前线程名称:ForkJoinPool.commonPool-worker-1
java.util.concurrent.CompletionException: java.lang.RuntimeException
 at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
 at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
 at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592)
 at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582)
 at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
 at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
 at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
 at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.lang.RuntimeException
 at cn.eovie.future.FutureWhenTest.lambda$main$0(FutureWhenTest.java:13)
 at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
 ... 5 more
你的程序异常啦

5. whenComplete方法

CompletableFuture的whenComplete方法表示,某个任务执行完成后,执行的回调方法,无返回值;并且whenComplete方法返回的CompletableFuture的result是上个任务的结果

public class FutureWhenTest 

    public static void main(String[] args) throws ExecutionException, InterruptedException 
        CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
                ()->
                    System.out.println("当前线程名称:" + Thread.currentThread().getName());
                    try 
                        Thread.sleep(2000L);
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                    return "技术人成长之路";
                
        );

        CompletableFuture<String> rstFuture = orgFuture.whenComplete((a, throwable) -> 
            System.out.println("当前线程名称:" + Thread.currentThread().getName());
            System.out.println("上个任务执行完啦,还把" + a + "传过来");
            if ("技术人成长之路".equals(a)) 
                System.out.println("666");
            
            System.out.println("233333");
        );

        System.out.println(rstFuture.get());
    


//输出
当前线程名称:ForkJoinPool.commonPool-worker-1
当前线程名称:ForkJoinPool.commonPool-worker-1
上个任务执行完啦,还把技术人成长之路传过来
666
233333
技术人成长之路

6. handle方法

CompletableFuture的handle方法表示,某个任务执行完成后,执行回调方法,并且是有返回值的;并且handle方法返回的CompletableFuture的result是回调方法执行的结果。

public class FutureHandlerTest 

    public static void main(String[] args) throws ExecutionException, InterruptedException 
        CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(
                (以上是关于Java 8 的异步编程利器 CompletableFuture 详解的主要内容,如果未能解决你的问题,请参考以下文章

Java 8 的异步编程利器 CompletableFuture 详解

(94) 组合式异步编程 / 计算机程序的思维逻辑

Java8 异步编程利器 CompletableFuture 详解(全网看这一篇就行)

异步编程利器:CompletableFuture如何操作?

Java基础学习总结(198)—— 异步编程利器 CompletableFuture 使用详解

Java 8 的异步利器:CompletableFuture源码级解析(建议精读)