文章详情

登录

短信预约-IT技能 免费直播动态提醒

选择考试省份

  • 北京
  • 上海
  • 天津
  • 重庆
  • 河北
  • 山东
  • 辽宁
  • 黑龙江
  • 吉林
  • 甘肃
  • 青海
  • 河南
  • 江苏
  • 湖北
  • 湖南
  • 江西
  • 浙江
  • 广东
  • 云南
  • 福建
  • 海南
  • 山西
  • 四川
  • 陕西
  • 贵州
  • 安徽
  • 广西
  • 内蒙
  • 西藏
  • 新疆
  • 宁夏
  • 兵团

请输入下面的图形验证码

提交验证

短信预约提醒成功

CompletableFuture使用详解

2023-08-16 17:46

关注

一、简介

1.1 概述

在上一篇文章《CompletionService使用与源码分析》中，已经介绍过了Future的局限性，它没法直接对多个任务进行链式、组合等处理，需要借助并发工具类才能完成，实现逻辑比较复杂。

而CompletableFuture是对Future的扩展和增强。CompletableFuture实现了Future接口，并在此基础上进行了丰富的扩展，完美弥补了Future的局限性，同时CompletableFuture实现了对任务编排的能力。借助这项能力，可以轻松地组织不同任务的运行顺序、规则以及方式。从某种程度上说，这项能力是它的核心能力。而在以往，虽然通过CountDownLatch等工具类也可以实现任务的编排，但需要复杂的逻辑处理，不仅耗费精力且难以维护。

CompletableFuture的继承结构如下：
在这里插入图片描述

CompletionStage接口定义了任务编排的方法，执行某一阶段，可以向下执行后续阶段。异步执行的，默认线程池是ForkJoinPool.commonPool()，但为了业务之间互不影响，且便于定位问题，强烈推荐使用自定义线程池。

CompletableFuture中默认线程池如下：

// 根据commonPool的并行度来选择,而并行度的计算是在ForkJoinPool的静态代码段完成的private static final boolean useCommonPool =    (ForkJoinPool.getCommonPoolParallelism() > 1);private static final Executor asyncPool = useCommonPool ?    ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

ForkJoinPool中初始化commonPool的参数

static {    // initialize field offsets for CAS etc    try {        U = sun.misc.Unsafe.getUnsafe();        Class k = ForkJoinPool.class;        CTL = U.objectFieldOffset            (k.getDeclaredField("ctl"));        RUNSTATE = U.objectFieldOffset            (k.getDeclaredField("runState"));        STEALCOUNTER = U.objectFieldOffset            (k.getDeclaredField("stealCounter"));        Class tk = Thread.class;        ……    } catch (Exception e) {        throw new Error(e);    }    commonMaxSpares = DEFAULT_COMMON_MAX_SPARES;    defaultForkJoinWorkerThreadFactory =        new DefaultForkJoinWorkerThreadFactory();    modifyThreadPermission = new RuntimePermission("modifyThread");    // 调用makeCommonPool方法创建commonPool,其中并行度为逻辑核数-1    common = java.security.AccessController.doPrivileged        (new java.security.PrivilegedAction() {            public ForkJoinPool run() { return makeCommonPool(); }});    int par = common.config & SMASK; // report 1 even if threads disabled    commonParallelism = par > 0 ? par : 1;}

1.2 功能

1.2.1 常用方法

依赖关系
and集合关系
or聚合关系
并行执行
结果处理

1.2.2 异步操作

CompletableFuture提供了四个静态方法来创建一个异步操作：

public static CompletableFuture runAsync(Runnable runnable)public static CompletableFuture runAsync(Runnable runnable, Executor executor)public static  CompletableFuture supplyAsync(Supplier supplier)public static  CompletableFuture supplyAsync(Supplier supplier, Executor executor)

这四个方法的区别：

  • runAsync() 以Runnable函数式接口类型为参数，没有返回结果，supplyAsync() 以Supplier函数式接口类型为参数，返回结果类型为U；Supplier接口的 get()是有返回值的(会阻塞)
  • 使用没有指定Executor的方法时，内部使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。
  • 默认情况下CompletableFuture会使用公共的ForkJoinPool线程池，这个线程池默认创建的线程数是 CPU 的核数（也可以通过 JVM option:-Djava.util.concurrent.ForkJoinPool.common.parallelism 来设置ForkJoinPool线程池的线程数）。如果所有CompletableFuture共享一个线程池，那么一旦有任务执行一些很慢的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成线程饥饿，进而影响整个系统的性能。所以，强烈建议你要根据不同的业务类型创建不同的线程池，以避免互相干扰

异步操作

Runnable runnable = () -> System.out.println("无返回结果异步任务");CompletableFuture.runAsync(runnable);CompletableFuture future = CompletableFuture.supplyAsync(() -> {    System.out.println("有返回值的异步任务");    try {        Thread.sleep(5000);    } catch (InterruptedException e) {        e.printStackTrace();    }    return "Hello World";});String result = future.get();

获取结果(join&get)

join()和get()方法都是用来获取CompletableFuture异步之后的返回值。join()方法抛出的是uncheck异常（即未经检查的异常),不会强制开发者抛出。get()方法抛出的是经过检查的异常，ExecutionException, InterruptedException 需要用户手动处理（抛出或者 try catch）

结果处理

当CompletableFuture的计算结果完成，或者抛出异常的时候，我们可以执行特定的 Action。主要是下面的方法：

public CompletableFuture whenComplete(BiConsumer action)public CompletableFuture whenCompleteAsync(BiConsumer action)public CompletableFuture whenCompleteAsync(BiConsumer action, Executor executor)
  • Action的类型是BiConsumer，它可以处理正常的计算结果，或者异常情况。
  • 方法不以Async结尾，意味着Action使用相同的线程执行，而Async可能会使用其它的线程去执行(如果使用相同的线程池，也可能会被同一个线程选中执行)。
  • 这几个方法都会返回CompletableFuture，当Action执行完毕后它的结果返回原始的CompletableFuture的计算结果或者返回异常
CompletableFuture future = CompletableFuture.supplyAsync(() -> {    try {        TimeUnit.SECONDS.sleep(1);    } catch (InterruptedException e) {    }    if (new Random().nextInt(10) % 2 == 0) {        int i = 12 / 0;    }    System.out.println("执行结束！");    return "test";});// 任务完成或异常方法完成时执行该方法// 如果出现了异常,任务结果为nullfuture.whenComplete(new BiConsumer() {    @Override    public void accept(String t, Throwable action) {        System.out.println(t+" 执行完成！");    }});// 出现异常时先执行该方法future.exceptionally(new Function() {    @Override    public String apply(Throwable t) {        System.out.println("执行失败：" + t.getMessage());        return "异常xxxx";    }});future.get();

上面的代码当出现异常时，输出结果如下

执行失败：java.lang.ArithmeticException: / by zeronull 执行完成！

二、应用场景

2.1 结果转换

将上一段任务的执行结果作为下一阶段任务的入参参与重新计算，产生新的结果。

thenApply

thenApply接收一个函数作为参数，使用该函数处理上一个CompletableFuture调用的结果，并返回一个具有处理结果的Future对象。

常用使用：

public  CompletableFuture thenApply(Function fn)public  CompletableFuture thenApplyAsync(Function fn)

具体使用：

CompletableFuture future = CompletableFuture.supplyAsync(() -> {    int result = 100;    System.out.println("第一次运算：" + result);    return result;}).thenApply(number -> {    int result = number * 3;    System.out.println("第二次运算：" + result);    return result;});

thenCompose

thenCompose的参数为一个返回CompletableFuture实例的函数，该函数的参数是先前计算步骤的结果。

常用方法：

public  CompletableFuture thenCompose(Function> fn);public  CompletableFuture thenComposeAsync(Function> fn) ;

具体使用：

CompletableFuture future = CompletableFuture    .supplyAsync(new Supplier() {        @Override        public Integer get() {            int number = new Random().nextInt(30);            System.out.println("第一次运算：" + number);            return number;        }    })    .thenCompose(new Function>() {        @Override        public CompletionStage apply(Integer param) {            return CompletableFuture.supplyAsync(new Supplier() {                @Override                public Integer get() {                    int number = param * 2;                    System.out.println("第二次运算：" + number);                    return number;                }            });        }    });

thenApply 和 thenCompose的区别：

  • thenApply转换的是泛型中的类型，返回的是同一个CompletableFuture；
  • thenCompose将内部的CompletableFuture调用展开来并使用上一个CompletableFutre调用的结果在下一步的CompletableFuture调用中进行运算，是生成一个新的CompletableFuture。

2.2 结果消费

与结果处理和结果转换系列函数返回一个新的CompletableFuture不同，结果消费系列函数只对结果执行Action，而不返回新的计算值。

根据对结果的处理方式，结果消费函数又可以分为下面三大类：

  • thenAccept()：对单个结果进行消费
  • thenAcceptBoth()：对两个结果进行消费
  • thenRun()：不关心结果，只对结果执行Action

thenAccept

观察该系列函数的参数类型可知，它们是函数式接口Consumer，这个接口只有输入，没有返回值。

常用方法：

public CompletionStage thenAccept(Consumer action);public CompletionStage thenAcceptAsync(Consumer action);

具体使用：

CompletableFuture future = CompletableFuture    .supplyAsync(() -> {        int number = new Random().nextInt(10);        System.out.println("第一次运算：" + number);        return number;    }).thenAccept(number ->                  System.out.println("第二次运算：" + number * 5));

thenAcceptBoth

thenAcceptBoth函数的作用是，当两个CompletionStage都正常完成计算的时候，就会执行提供的action消费两个异步的结果。

常用方法：

public  CompletionStage thenAcceptBoth(CompletionStage other,BiConsumer action);public  CompletionStage thenAcceptBothAsync(CompletionStage other,BiConsumer action);

具体使用：

CompletableFuture futrue1 = CompletableFuture.supplyAsync(new Supplier() {    @Override    public Integer get() {        int number = new Random().nextInt(3) + 1;        try {            TimeUnit.SECONDS.sleep(number);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("任务1结果：" + number);        return number;    }});CompletableFuture future2 = CompletableFuture.supplyAsync(new Supplier() {    @Override    public Integer get() {        int number = new Random().nextInt(3) + 1;        try {            TimeUnit.SECONDS.sleep(number);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("任务2结果：" + number);        return number;    }});futrue1.thenAcceptBoth(future2, new BiConsumer() {    @Override    public void accept(Integer x, Integer y) {        System.out.println("最终结果：" + (x + y));    }});

thenRun

thenRun也是对线程任务结果的一种消费函数，与thenAccept不同的是，thenRun会在上一阶段 CompletableFuture计算完成的时候执行一个Runnable，而Runnable并不使用该CompletableFuture计算的结果。

常用方法：

public CompletionStage thenRun(Runnable action);public CompletionStage thenRunAsync(Runnable action);

具体使用：

CompletableFuture future = CompletableFuture.supplyAsync(() -> {    int number = new Random().nextInt(10);    System.out.println("第一阶段：" + number);    return number;}).thenRun(() ->           System.out.println("thenRun 执行"));

2.3 结果组合

thenCombine

合并两个线程任务的结果，并进一步处理。

常用方法：

public  CompletableFuture thenCombine(CompletionStage other,BiFunction fn);public  CompletableFuture thenCombineAsync(CompletionStage other,BiFunction fn);public  CompletableFuture thenCombineAsync(CompletionStage other,BiFunction fn, Executor executor);

具体使用：

CompletableFuture future1 = CompletableFuture    .supplyAsync(new Supplier() {        @Override        public Integer get() {            int number = new Random().nextInt(10);            System.out.println("任务1结果：" + number);            return number;        }    });CompletableFuture future2 = CompletableFuture    .supplyAsync(new Supplier() {        @Override        public Integer get() {            int number = new Random().nextInt(10);            System.out.println("任务2结果：" + number);            return number;        }    });CompletableFuture result = future1    .thenCombine(future2, new BiFunction() {        @Override        public Integer apply(Integer x, Integer y) {            return x + y;        }    });System.out.println("组合后结果：" + result.get());

2.4 任务交互

线程交互指将两个线程任务获取结果的速度相比较，按一定的规则进行下一步处理。

applyToEither

两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的转化操作。

常用方法：

public  CompletionStage applyToEither(CompletionStage other,Function fn);public  CompletionStage applyToEitherAsync(CompletionStage other,Function fn);

具体使用：

CompletableFuture future1 = CompletableFuture    .supplyAsync(new Supplier() {        @Override        public Integer get() {            int number = new Random().nextInt(10);            try {                TimeUnit.SECONDS.sleep(number);            } catch (InterruptedException e) {                e.printStackTrace();            }            System.out.println("任务1结果:" + number);            return number;        }    });CompletableFuture future2 = CompletableFuture.supplyAsync(new Supplier() {    @Override    public Integer get() {        int number = new Random().nextInt(10);        try {            TimeUnit.SECONDS.sleep(number);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("任务2结果:" + number);        return number;    }});future1.applyToEither(future2, new Function() {    @Override    public Integer apply(Integer number) {        System.out.println("最快结果：" + number);        return number * 2;    }});

acceptEither

两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的消费操作。

常用方法：

public CompletionStage acceptEither(CompletionStage other,Consumer action);public CompletionStage acceptEitherAsync(CompletionStage other,Consumer action);

具体使用：

CompletableFuture future1 = CompletableFuture.supplyAsync(new Supplier() {    @Override    public Integer get() {        int number = new Random().nextInt(10) + 1;        try {            TimeUnit.SECONDS.sleep(number);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("第一阶段：" + number);        return number;    }});CompletableFuture future2 = CompletableFuture.supplyAsync(new Supplier() {    @Override    public Integer get() {        int number = new Random().nextInt(10) + 1;        try {            TimeUnit.SECONDS.sleep(number);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("第二阶段：" + number);        return number;    }});future1.acceptEither(future2, new Consumer() {    @Override    public void accept(Integer number) {        System.out.println("最快结果：" + number);    }});

runAfterEither

两个线程任务相比较，有任何一个执行完成，就进行下一步操作，不关心运行结果。

常用方法：

public CompletionStage runAfterEither(CompletionStage other,Runnable action);public CompletionStage runAfterEitherAsync(CompletionStage other,Runnable action);

具体使用：

CompletableFuture future1 = CompletableFuture.supplyAsync(new Supplier() {    @Override    public Integer get() {        int number = new Random().nextInt(5);        try {            TimeUnit.SECONDS.sleep(number);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("任务1结果：" + number);        return number;    }});CompletableFuture future2 = CompletableFuture.supplyAsync(new Supplier() {    @Override    public Integer get() {        int number = new Random().nextInt(5);        try {            TimeUnit.SECONDS.sleep(number);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("任务2结果:" + number);        return number;    }});future1.runAfterEither(future2, new Runnable() {    @Override    public void run() {        System.out.println("已经有一个任务完成了");    }}).join();

anyOf

anyOf() 的参数是多个给定的 CompletableFuture，当其中的任何一个完成时，方法返回这个 CompletableFuture。

常用方法：

public static CompletableFuture anyOf(CompletableFuture... cfs) 
具体使用：
 
Random random = new Random();CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {    try {        TimeUnit.SECONDS.sleep(random.nextInt(5));    } catch (InterruptedException e) {        e.printStackTrace();    }    return "hello";});CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {    try {        TimeUnit.SECONDS.sleep(random.nextInt(1));    } catch (InterruptedException e) {        e.printStackTrace();    }    return "world";});CompletableFuture result = CompletableFuture.anyOf(future1, future2); 
allOf
 
allOf方法用来实现多 CompletableFuture 的同时返回。
 
常用方法：
 
public static CompletableFuture allOf(CompletableFuture... cfs)
 
具体使用：
 
CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {    try {        TimeUnit.SECONDS.sleep(2);    } catch (InterruptedException e) {        e.printStackTrace();    }    System.out.println("future1完成！");    return "future1完成！";});CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {    System.out.println("future2完成！");    return "future2完成！";});CompletableFuture combindFuture = CompletableFuture.allOf(future1, future2);try {    combindFuture.get();} catch (InterruptedException e) {    e.printStackTrace();} catch (ExecutionException e) {    e.printStackTrace();}
 
CompletableFuture常用方法总结:
 在这里插入图片描述
 注：CompletableFuture中还有很多功能丰富的方法，这里就不一一列举。
 
三、使用案例
 
实现最优的“烧水泡茶”程序
 
著名数学家华罗庚先生在《统筹方法》这篇文章里介绍了一个烧水泡茶的例子，文中提到最优的工序应该是下面这样：
  
对于烧水泡茶这个程序，一种最优的分工方案：用两个线程 T1 和 T2 来完成烧水泡茶程序，T1 负责洗水壶、烧开水、泡茶这三道工序，T2 负责洗茶壶、洗茶杯、拿茶叶三道工序，其中 T1 在执行泡茶这道工序时需要等待 T2 完成拿茶叶的工序。
 
基于Future实现
 
public class FutureTaskTest{    public static void main(String[] args) throws ExecutionException, InterruptedException {        // 创建任务T2的FutureTask        FutureTask ft2 = new FutureTask<>(new T2Task());        // 创建任务T1的FutureTask        FutureTask ft1 = new FutureTask<>(new T1Task(ft2));        // 线程T1执行任务ft2        Thread T1 = new Thread(ft2);        T1.start();        // 线程T2执行任务ft1        Thread T2 = new Thread(ft1);        T2.start();        // 等待线程T1执行结果        System.out.println(ft1.get());    }}// T1Task需要执行的任务：// 洗水壶、烧开水、泡茶class T1Task implements Callable {    FutureTask ft2;    // T1任务需要T2任务的FutureTask    T1Task(FutureTask ft2){        this.ft2 = ft2;    }    @Override    public String call() throws Exception {        System.out.println("T1:洗水壶...");        TimeUnit.SECONDS.sleep(1);        System.out.println("T1:烧开水...");        TimeUnit.SECONDS.sleep(15);        // 获取T2线程的茶叶        String tf = ft2.get();        System.out.println("T1:拿到茶叶:"+tf);        System.out.println("T1:泡茶...");        return "上茶:" + tf;    }}// T2Task需要执行的任务:// 洗茶壶、洗茶杯、拿茶叶class T2Task implements Callable {    @Override    public String call() throws Exception {        System.out.println("T2:洗茶壶...");        TimeUnit.SECONDS.sleep(1);        System.out.println("T2:洗茶杯...");        TimeUnit.SECONDS.sleep(2);        System.out.println("T2:拿茶叶...");        TimeUnit.SECONDS.sleep(1);        return "龙井";    }}
 
基于CompletableFuture实现
 
public class CompletableFutureTest {    public static void main(String[] args) {        //任务1：洗水壶->烧开水        CompletableFuture f1 = CompletableFuture            .runAsync(() -> {                System.out.println("T1:洗水壶...");                sleep(1, TimeUnit.SECONDS);                System.out.println("T1:烧开水...");                sleep(15, TimeUnit.SECONDS);            });        //任务2：洗茶壶->洗茶杯->拿茶叶        CompletableFuture f2 = CompletableFuture            .supplyAsync(() -> {                System.out.println("T2:洗茶壶...");                sleep(1, TimeUnit.SECONDS);                System.out.println("T2:洗茶杯...");                sleep(2, TimeUnit.SECONDS);                System.out.println("T2:拿茶叶...");                sleep(1, TimeUnit.SECONDS);                return "龙井";            });        //任务3：任务1和任务2完成后执行：泡茶        CompletableFuture f3 = f1.thenCombine(f2, (__, tf) -> {            System.out.println("T1:拿到茶叶:" + tf);            System.out.println("T1:泡茶...");            return "上茶:" + tf;        });        //等待任务3执行结果        System.out.println(f3.join());    }    static void sleep(int t, TimeUnit u){        try {            u.sleep(t);        } catch (InterruptedException e) {        }    }}
                
来源地址：https://blog.csdn.net/sermonlizhi/article/details/123356877                        
阅读原文内容投诉
免责声明：
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。 
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
软考中级精品资料免费领 
  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 资料下载
  • 历年真题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析（完整版）
     难度      813人已做 
     查看 
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析
     难度      354人已做 
     查看 
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析
     难度      318人已做 
     查看 
  • 2024年上半年软考高项第一、二批次真题考点汇总（完整版）
     难度      435人已做 
     查看 
  • 2024年上半年系统架构设计师考试综合知识真题
     难度      224人已做 
     查看 
相关文章
发现更多好内容
猜你喜欢
AI推送时光机
CompletableFuture使用详解
后端开发2023-08-16
Java CompletableFuture的使用详解
后端开发2024-04-02
详解Java8中CompletableFuture类的使用
后端开发2023-05-15
CompletableFuture的使用
后端开发2023-09-01
CompletableFuture使用详解（全网看这一篇就行）
后端开发2023-09-12
详解Java CompletableFuture使用方法以及与FutureTask的区别
后端开发2024-04-02
详解Java8 CompletableFuture的并行处理用法
后端开发2024-04-02
怎么使用CompletableFuture
后端开发2023-06-15
CompletableFuture怎么使用
后端开发2023-07-06
CompletableFuture 异步编排示例详解
后端开发2024-04-02
Java CompletableFuture实现原理分析详解
后端开发2024-04-02
Java8中CompletableFuture怎么使用
后端开发2023-06-29
Java 8中CompletableFuture如何使用
后端开发2023-06-15
Java多线程开发工具之CompletableFuture的应用详解
后端开发2023-03-20
Java8 使用CompletableFuture 构建异步应用方式
后端开发2024-04-02
Java8并发新特性CompletableFuture怎么使用
后端开发2023-06-30
Java8中CompletableFuture的用法全解
后端开发2024-04-02
Java8中的CompletableFuture类怎么使用
后端开发2023-07-06
Java8如何使用CompletableFuture构建异步应用方式
后端开发2023-06-25
详解Java如何在CompletableFuture中实现日志记录
后端开发2023-05-16
 位置：首页-资讯-后端开发
 咦！没有更多了？去看看其它编程学习网 内容吧 
首页课程
资料下载问答资讯