程序员子龙(Java面试 + Java学习) 程序员子龙(Java面试 + Java学习)
首页
学习指南
工具
开源项目
技术书籍

程序员子龙

Java 开发从业者
首页
学习指南
工具
开源项目
技术书籍
  • 基础

  • JVM

  • Spring

  • 并发编程

    • 并发和并行
    • 什么是多线程
    • Java 并发容器有哪些?
    • Java的Future机制详解
      • 什么是AQS
      • 一文搞懂 ThreadLocal
      • java 阻塞队列 详细介绍
      • java线程池使用最全详解
      • 面试官问我什么是JMM
      • CountDownLatch、Semaphore和CyclicBarrier
      • Java:线程的六种状态及转化
    • Mybatis

    • 网络编程

    • 数据库

    • 缓存

    • 设计模式

    • 分布式

    • 高并发

    • SpringBoot

    • SpringCloudAlibaba

    • Nginx

    • 面试

    • 生产问题

    • 系统设计

    • 消息中间件

    • Java
    • 并发编程
    程序员子龙
    2024-01-29
    目录

    Java的Future机制详解

    # 类图

    # Future 主要功能

    Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果,该方法会阻塞直到任务返回结果。

    • boolean cancel (boolean mayInterruptIfRunning) 取消任务的执行。参数指定是否立即中断任务执行,或者等等任务结束
    • boolean isCancelled () 任务是否已经取消,任务正常完成前将其取消,则返回 true
    • boolean isDone () 任务是否已经完成。需要注意的是如果任务正常终止、异常或取消,都将返回true
    • V get () throws InterruptedException, ExecutionException 等待任务执行结束,然后获得V类型的结果。InterruptedException 线程被中断异常, ExecutionException任务执行异常,如果任务被取消,还会抛出CancellationException
    • V get (long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException 同上面的get功能一样,多了设置超时时间。参数timeout指定超时时间,uint指定时间的单位,在枚举类TimeUnit中有相关的定义。如果计算超时,将抛出TimeoutException

    # Future的应用场景

    在查询商品信息(包括商品基本信息、商品价格、商品库存、商品图片、商品销售状态等)。这些信息分布在不同的业务中心,由不同的系统提供服务。如果采用同步方式,假设一个接口需要50ms,那么一个商品查询下来就需要200ms-300ms,这对于我们来说是不满意的。如果使用Future改造则需要的就是最长耗时服务的接口,也就是50ms左右。

    # Future使用

    利用 FutureTask 创建 Future

    Future实际采用FutureTask实现,该对象相当于是消费者和生产者的桥梁,消费者通过 FutureTask 存储任务的处理结果,更新任务的状态:未开始、正在处理、已完成等。而生产者拿到的 FutureTask 被转型为 Future 接口,可以阻塞式获取任务的处理结果,非阻塞式获取任务处理状态。

    public class FutureTaskDemo {
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            Task task = new Task();
            //构建futureTask
            FutureTask<Integer> futureTask = new FutureTask<>(task);
            //作为Runnable入参
            new Thread(futureTask).start();
    
            System.out.println("主程序继续执行。。");
    
            System.out.println("task运行结果:"+futureTask.get());
        }
    
        static class Task implements Callable<Integer> {
    
            @Override
            public Integer call() throws Exception {
                System.out.println("子线程正在计算");
                int sum = 0;
                for (int i = 0; i < 1000; i++) {
                    sum += i;
                }
                // 模拟业务执行时间
                Thread.sleep(1000);
                return sum;
            }
        }
    }
    
    主程序继续执行。。
    子线程正在计算
    task运行结果:499500
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33

    # Future的局限性

    从本质上说,Future表示一个异步计算的结果。它提供了isDone()来检测计算是否已经完成,并且在计算结束后,可以通过get()方法来获取计算结果。在异步计算中,Future确实是个非常优秀的接口。但是,它的本身也确实存在着许多限制:

    • 并发执行多任务:Future只提供了get()方法来获取结果,并且是阻塞的。所以,除了等待你别无他法;
    • 无法对多个任务进行链式调用:如果你希望在计算任务完成后执行特定动作,比如发邮件,但Future却没有提供这样的能力;
    • 无法组合多个任务:如果你运行了10个任务,并期望在它们全部执行结束后执行特定动作,那么在Future中这是无能为力的;
    • 没有异常处理:Future接口中没有关于异常处理的方法;

    # CompleteableFuture

    简单的任务,用Future获取结果还好,但我们并行提交的多个异步任务,往往并不是独立的,很多时候业务逻辑处理存在串行[依赖]、并行、聚合的关系。如果要我们手动用 Fueture 实现,是非常麻烦的。

    JDK1.8 才新加入的一个实现类 CompletableFuture,实现了 Future<T>,CompletionStage<T>两个接口。实现了 Future 接口,意味着可以像以前一样通过阻塞或者轮询的方式获得结果。

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

    CompletionStage接口: 执行某一个阶段,可向下执行后续阶段。异步执行,默认线程池是ForkJoinPool.commonPool()。它代表

    了一个特定的计算的阶段,可以同步或者异步的被完成。你可以把它看成一个计算流水线上的一个单元,并最终会产生一个最终结果,这意味着几个CompletionStage 可以串联起来,一个完成的阶段可以触发下一阶段的执行,接着触发下一次,再接着触发下一次,……….。

    创建异步操作

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

    public static CompletableFuture<Void> runAsync(Runnable runnable)
    public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
    
    1
    2
    3
    4

    这四个方法区别在于:

    • runAsync 方法以Runnable函数式接口类型为参数,没有返回结果,supplyAsync 方法Supplier函数式接口类型为参数,返回结果类型为U;Supplier 接口的 get() 方法是有返回值的(会阻塞),
    • Asynsc表示异步,而supplyAsync与runAsync 不同在与前者异步返回一个结果, 后者是 void.
    • 没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池,则使用指定的线程池运行。
    • 默认情况下 CompletableFuture 会使用公共的 ForkJoinPool 线程池,这个线程池默认创建的线程数是 CPU 的核数(也可以通过 JVM option:-Djava.util.concurrent.ForkJoinPool.common.parallelism 来设置 ForkJoinPool 线程池的线程数)。如果所有 CompletableFuture 共享一个线程池,那么一旦有任务执行一些很慢的 I/O 操作,就会导致线程池中所有线程都阻塞在 I/O 操作上,从而造成线程饥饿,进而影响整个系统的性能。所以,强烈建议你要根据不同的业务类型创建不同的线程池,以避免互相干扰。
    public static void main(String[] args) throws ExecutionException, InterruptedException {
    
        Runnable runnable = () -> System.out.println("执行无返回结果的异步任务");
        CompletableFuture.runAsync(runnable);
    
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行有返回值的异步任务");
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "你好 2022";
        });
        String result = future.join();
        System.out.println(result);
    
    }
    
    执行无返回结果的异步任务
    执行有返回值的异步任务
    你好 2022
    
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23

    获取结果

    join&get

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

    public T get() 
    
    public T get(long timeout, TimeUnit unit) 
    
    public T getNow(T valueIfAbsent) 
    
    public T join() 
    
    1
    2
    3
    4
    5
    6
    7

    getNow 有点特殊,如果结果已经计算完则返回结果或者抛出异常,否则返回给定的 valueIfAbsent 值。

    下面这段代码编译时候不会报错

    public static void main(String[] args) {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(()->{
           int i = 1/0;
           return 100;
        });
    
        future.join();
    }
    
    1
    2
    3
    4
    5
    6
    7
    8

    结果处理

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

    public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action) 
    public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action) 
    public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)
    
    1
    2
    3
    • Action的类型是BiConsumer,它可以处理正常的计算结果,或者异常情况。
    • 方法不以Async结尾,意味着Action使用相同的线程执行,而Async可能会使用其它的线程去执行(如果使用相同的线程池,也可能会被同一个线程选中执行)。
    • 这几个方法都会返回CompletableFuture,当Action执行完毕后它的结果返回原始的CompletableFuture的计算结果或者返回异常
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
    
            CompletableFuture<String> 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";
            });
    
            future.whenComplete(new BiConsumer<String, Throwable>() {
                @Override
                public void accept(String t, Throwable action) {
                    System.out.println(t+" 执行完成!");
                }
            });
    
            // 异常时执行
            future.exceptionally(new Function<Throwable, String>() {
                @Override
                public String apply(Throwable t) {
                    System.out.println("执行失败:" + t.getMessage());
                    return "异常xxxx";
                }
            }).join();
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31

    执行结束!

    test 执行完成!

    或者

    执行失败:java.lang.ArithmeticException: / by zero

    null 执行完成!

    结果转换

    所谓结果转换,就是将上一段任务的执行结果作为下一阶段任务的入参参与计算,产生新的结果。

    thenApply

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

    public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
    public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
    
    1
    2
      CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
                int result = 100;
                System.out.println("一阶段:" + result);
                return result;
            }).thenApply(number -> {
                int result = number * 3;
                System.out.println("二阶段:" + result);
                return result;
            });
    
    1
    2
    3
    4
    5
    6
    7
    8
    9

    一阶段:100 二阶段:300

    thenCompose

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

    public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn); 
    public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) ;
    
    
    1
    2
    3
    
        public static void main(String[] args) {
            CompletableFuture<Integer> future = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @Override
                        public Integer get() {
                            int number = new Random().nextInt(30);
                            System.out.println("第一阶段:" + number);
                            return number;
                        }
                    })
                    .thenCompose(new Function<Integer, CompletionStage<Integer>>() {
                        @Override
                        public CompletionStage<Integer> apply(Integer param) {
                            return CompletableFuture.supplyAsync(new Supplier<Integer>() {
                                @Override
                                public Integer get() {
                                    int number = param * 2;
                                    System.out.println("第二阶段:" + number);
                                    return number;
                                }
                            });
                        }
                    });
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25

    第一阶段:5 第二阶段:10

    thenApply 和 thenCompose的区别

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

    结果消费

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

    根据对结果的处理方式,结果消费函数又分为:

    • thenAccept系列:对单个结果进行消费
    • thenAcceptBoth系列:对两个结果进行消费
    • thenRun系列:不关心结果,只对结果执行Action

    thenAccept

    通过观察该系列函数的参数类型可知,它们是函数式接口Consumer,这个接口只有输入,没有返回值,它的入参是上一个阶段计算后的结果。

    public CompletionStage<Void> thenAccept(Consumer<? super T> action);
    public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
    
    1
    2
      public static void main(String[] args) {
            CompletableFuture<Void> future = CompletableFuture
                    .supplyAsync(() -> {
                        int number = new Random().nextInt(10);
                        System.out.println("第一阶段:" + number);
                        return number;
                    }).thenAccept(number ->
                            System.out.println("第二阶段:" + number * 5));
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9

    第一阶段:8 第二阶段:40

    thenAcceptBoth

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

    public <U> CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action);
    public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action);
    
    1
    2
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
                @Override
                public Integer get() {
                    int number = new Random().nextInt(3) + 1;
                    try {
                        TimeUnit.SECONDS.sleep(number);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("第一阶段:" + number);
                    return number;
                }
            });
    
            CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
                @Override
                public Integer get() {
                    int number = new Random().nextInt(3) + 1;
                    try {
                        TimeUnit.SECONDS.sleep(number);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("第二阶段:" + number);
                    return number;
                }
            });
    
            futrue1.thenAcceptBoth(future2, new BiConsumer<Integer, Integer>() {
                @Override
                public void accept(Integer x, Integer y) {
                    System.out.println("最终结果:" + (x + y));
                }
            });
    
            future2.get();
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38

    第一阶段:2 第二阶段:3 最终结果:5

    thenRun

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

    public CompletionStage<Void> thenRun(Runnable action);
    public CompletionStage<Void> thenRunAsync(Runnable action);
    
    1
    2
        public static void main(String[] args) {
    
            CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
                int number = new Random().nextInt(10);
                System.out.println("第一阶段:" + number);
                return number;
            }).thenRun(new Runnable() {
                @Override
                public void run() {
                    System.out.println("thenRun 执行");
                }
            });
    
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14

    第一阶段:2 thenRun 执行

    结果组合

    thenCombine

    thenCombine 方法,合并两个线程任务的结果,并进一步处理。

    public <U,V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
    public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
    
    1
    2
        public static void main(String[] args) {
            CompletableFuture<Integer> future1 = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @Override
                        public Integer get() {
                            int number = new Random().nextInt(10);
                            System.out.println("第一阶段:" + number);
                            return number;
                        }
                    });
            CompletableFuture<Integer> future2 = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @Override
                        public Integer get() {
                            int number = new Random().nextInt(10);
                            System.out.println("第二阶段:" + number);
                            return number;
                        }
                    });
            CompletableFuture<Integer> result = future1
                    .thenCombine(future2, new BiFunction<Integer, Integer, Integer>() {
                        @Override
                        public Integer apply(Integer x, Integer y) {
                            return x + y;
                        }
                    });
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27

    第一阶段:0 第二阶段:1

    任务交互

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

    applyToEither

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

    public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T> other,Function<? super T, U> fn); public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn);            
    
    1
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            CompletableFuture<Integer> future1 = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @Override
                        public Integer get() {
                            int number = new Random().nextInt(10);
                            System.out.println("第一阶段start:" + number);
                            try {
                                TimeUnit.SECONDS.sleep(number);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            System.out.println("第一阶段end:" + number);
                            return number;
                        }
                    });
            CompletableFuture<Integer> future2 = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @Override
                        public Integer get() {
                            int number = new Random().nextInt(10);
                            System.out.println("第二阶段start:" + number);
                            try {
                                TimeUnit.SECONDS.sleep(number);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            System.out.println("第二阶段end:" + number);
                            return number;
                        }
                    });
    
            future1.applyToEither(future2, new Function<Integer, Integer>() {
                @Override
                public Integer apply(Integer number) {
                    System.out.println("最快结果:" + number);
                    return number * 2;
                }
            });
            future1.get();
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41

    第一阶段start:7 第二阶段start:1 第二阶段end:1 最快结果:1 第一阶段end:7

    acceptEither

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

    public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other,Consumer<? super T> action);
    public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action);
    
    1
    2
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            CompletableFuture<Integer> future1 = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @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<Integer> future2 = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @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<Integer>() {
                @Override
                public void accept(Integer number) {
                    System.out.println("最快结果:" + number);
                }
            });
            future1.get();
            future2.get();
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40

    第一阶段:2 最快结果:2 第二阶段:10

    runAfterEither

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

    public CompletionStage<Void> runAfterEither(CompletionStage<?> other,Runnable action);
    public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action);
    
    1
    2
        public static void main(String[] args) {
            CompletableFuture<Integer> future1 = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @Override
                        public Integer get() {
                            int number = new Random().nextInt(5);
                            try {
                                TimeUnit.SECONDS.sleep(number);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            System.out.println("第一阶段:" + number);
                            return number;
                        }
                    });
    
            CompletableFuture<Integer> future2 = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @Override
                        public Integer get() {
                            int number = new Random().nextInt(5);
                            try {
                                TimeUnit.SECONDS.sleep(number);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            System.out.println("第二阶段:" + number);
                            return number;
                        }
                    });
    
            future1.runAfterEither(future2, new Runnable() {
                @Override
                public void run() {
                    System.out.println("已经有一个任务完成了");
                }
            }).join();
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38

    第一阶段:1 已经有一个任务完成了

    runAfterBoth

    两个线程任务相比较,两个全部执行完成,才进行下一步操作,不关心运行结果。

    public CompletionStage<Void> runAfterBoth(CompletionStage<?> other,Runnable action);
    public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action);
    
    1
    2
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            CompletableFuture<Integer> future1 = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @Override
                        public Integer get() {
                            try {
                                TimeUnit.SECONDS.sleep(1);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            System.out.println("第一阶段:1");
                            return 1;
                        }
                    });
    
            CompletableFuture<Integer> future2 = CompletableFuture
                    .supplyAsync(new Supplier<Integer>() {
                        @Override
                        public Integer get() {
                            try {
                                TimeUnit.SECONDS.sleep(2);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            System.out.println("第二阶段:2");
                            return 2;
                        }
                    });
    
            future1.runAfterBoth(future2, new Runnable() {
                @Override
                public void run() {
                    System.out.println("上面两个任务都执行完成了。");
                }
            });
    
            future1.get();
            future2.get();
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39

    第一阶段:1 第二阶段:2 上面两个任务都执行完成了。

    anyOf

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

    allOf

    allOf方法用来实现多 CompletableFuture 的同时返回。

    public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)
    public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
    
    1
    2
     public static void main(String[] args) throws ExecutionException, InterruptedException {
            Random rand = new Random();
            CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(1000 + rand.nextInt(1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("future1完成");
                return 100;
            });
            CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(2000 + rand.nextInt(1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("future2完成");
                return "abc";
            });
            CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(3000 + rand.nextInt(1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("future3完成");
                return "123abc";
            });
            CompletableFuture.allOf(future1,future2,future3).thenRun(()->{
                System.out.println("All done!");
            });
    
            CompletableFuture<Object> f = CompletableFuture.anyOf(future1,future2,future3);
            System.out.println("结果:"+f.get());
    
            TimeUnit.SECONDS.sleep(5);
        }
    
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38

    future1完成 结果:100 future2完成 future3完成 All done!

    CompletableFuture常用方法总结

    image-20220102181655037

    应用场景

    描述依赖关系:

    1. thenApply() 把前面异步任务的结果,交给后面的Function
    2. thenCompose()用来连接两个有依赖关系的任务,结果由第二个任务返回

    描述and聚合关系:

    1. thenCombine:任务合并,有返回值
    2. thenAccepetBoth:两个任务执行完成后,将结果交给thenAccepetBoth消耗,无返回值。
    3. runAfterBoth:两个任务都执行完成后,执行下一步操作(Runnable)。

    描述or聚合关系:

    1. applyToEither:两个任务谁执行的快,就使用那一个结果,有返回值。
    2. acceptEither: 两个任务谁执行的快,就消耗那一个结果,无返回值。
    3. runAfterEither: 任意一个任务执行完成,进行下一步操作(Runnable)。

    并行执行:

    CompletableFuture类自己也提供了anyOf()和allOf()用于支持多个CompletableFuture并行执行

    总结 CompletableFuture 几个关键点:

    1、计算可以由 Future ,Consumer 或者 Runnable 接口中的 apply,accept或者 run 等方法表示。

    2、计算的执行主要有以下

    a. 默认执行

    b. 使用默认的 CompletionStage 的异步执行提供者异步执行。这些方法名使用 someActionAsync 这种格式表示。

    c. 使用 Executor 提供者异步执行。这些方法同样也是 someActionAsync 这 种格式,但是会增加一个 Executor 参数。

    CompletableFuture的API非常丰富,不用全部掌握,大概了解有哪些功能,使用时会查API就行

    上次更新: 2024/03/11, 15:54:57
    Java 并发容器有哪些?
    什么是AQS

    ← Java 并发容器有哪些? 什么是AQS→

    最近更新
    01
    一个注解,优雅的实现接口幂等性
    11-17
    02
    MySQL事务(超详细!!!)
    10-14
    03
    阿里二面:Kafka中如何保证消息的顺序性?这周被问到两次了
    10-09
    更多文章>
    Theme by Vdoing | Copyright © 2024-2024

        辽ICP备2023001503号-2

    • 跟随系统
    • 浅色模式
    • 深色模式
    • 阅读模式