CompletableFuture学习

本是古典何须时尚 2022-10-22 12:00 163阅读 0赞

### 文章目录 ###

*  CompletableFuture详解
 *   *  CompletableFuture的基本用法
     *  任务的异步运行
     *   *  异步执行Supplier
         *  异步执行Runnable类型的任务
     *  异步任务链
     *   *  thenApply ：以同步的方式继续处理上一个异步任务的结果
         *  thenApplyAsync: 以异步的方式继续处理上一个异步任务的结果
         *  thenAccept : 以同步的方式消费上一个异步任务的结果
         *  thenAcceptAsync 以异步的方式消费上一个异步任务的结果
         *  thenRun 以异步的而方式执行Runable任务
     *  合并多个Future
     *   *  thenCompose 方法示例
         *  thenCombine 方法示例
     *  多Future的并行计算
     *  CompletableFuture总结

# CompletableFuture详解 #

JDK1.8引入的新的Future，常用于异步编程之中，所谓异步编程，简单来说就是：“程序运算与应用程序主线程在不同的线程上完成，并且程序运算的线程能够向主线程通知其进度，以及成功与失败与否的非阻塞式编码方式”，

## CompletableFuture的基本用法 ##

CompletableFuture首先是一个Future。

public static void main(String[] args) {
        
            //double 类型的 CompletableFuture
            final CompletableFuture<Double> completableFuture = new CompletableFuture();
            //提交异步任务
            Executors.newCachedThreadPool().submit(() -> {
        
                try {
        
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println("finished");
                    //执行结束
                    completableFuture.complete(1234.5D);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            });
            //非阻塞获取异步任务的计算结果，和宁县，此刻异步任务未执行结束，那么可以采用默认值的方式
            //(该方法也可以被认为是放弃异步任务的执行结果，但不会取消异步任务的执行 )
            try {
        
                assert completableFuture.get() == 1234.5D;
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            } catch (ExecutionException e) {
        
                e.printStackTrace();
            }
        }

## 任务的异步运行 ##

CompletableFuture除了具备Future的基本特性之外，还可以直接使用它执行异步任务。通常情况下，任务的类型为Suppiler和Runnable，前者非常类似于Callable接口，可返回指定类型的运算结果，后者仍然只是关注异步任务运行本身。

### 异步执行Supplier ###

可以直接调用CompletableFuture的静态方法supplyAsync异步执行Supplier类型的任务。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 354);
    
    // 另一个重载 方法，运行传入 ExecutorService
    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 354, Executors.newCachedThreadPool());

### 异步执行Runnable类型的任务 ###

也可以直接调用CompletableFuture的静态方法runAsync 异步执行Runnable类型的任务。

CompletableFuture.runAsync(() -> {
        
        System.out.println("async task .");
    });

## 异步任务链 ##

CompletableFuture还允许将执行的异步任务结果继续交由下一级任务来执行，下一任务还可以有下一次，以此类推，这样就可以形成一个异步任务链或者任务pipeline。

### thenApply ：以同步的方式继续处理上一个异步任务的结果 ###

/**
         * 以同步的方式继续执行上一个异步任务的结果
         * supplyAsync 的 计算结果为 Java
         * thenApply 继续处理 “Java” ，返回字符串的长度
         * supplyAsync和thenApply 的任务执行是同一个线程
         */
        static void thenApply() throws ExecutionException, InterruptedException {
        
            final ExecutorService executor = Executors.newFixedThreadPool(3);
            final CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        
                System.out.println(Thread.currentThread().getName() + " supplyAsync");
                return "Java";
            }, executor).thenApply(e -> {
        
                System.out.println(Thread.currentThread().getName() + " thenApply");
                System.out.println(Thread.currentThread().getName() + " " + e);
                return e.length();
            });
            System.out.println(future.get());
        }

### thenApplyAsync: 以异步的方式继续处理上一个异步任务的结果 ###

/**
         * thenApplyAsync 的计算结果为“Java”
         * thenApplyAsync 继续处理“Java”，返回字符串的长度
         */
        static void thenApplyAsync() throws ExecutionException, InterruptedException {
        
            final ExecutorService executor = Executors.newFixedThreadPool(3);
            final CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        
                System.out.println(Thread.currentThread().getName() + " supplyAsync");
                return "Java";
            }, executor).thenApplyAsync(e -> {
        
                System.out.println(Thread.currentThread().getName() + " thenApplyAsync");
                System.out.println(Thread.currentThread().getName() + " " + e);
                return e.length();
            });
            System.out.println(future.get());
        }

### thenAccept : 以同步的方式消费上一个异步任务的结果 ###

/**
         * thenAccept : 以同步的方式消费上一个异步任务的结果
         */
        static void thenAccept() throws ExecutionException, InterruptedException {
        
            final ExecutorService executor = Executors.newFixedThreadPool(3);
            CompletableFuture.supplyAsync(() -> {
        
                System.out.println(Thread.currentThread().getName() + " supplyAsync");
                return "Java";
            }, executor).thenAccept(e -> {
        
                System.out.println(Thread.currentThread().getName() + " thenAccept");
                System.out.println(Thread.currentThread().getName() + " " + e);
            });
            executor.shutdown();
        }

### thenAcceptAsync 以异步的方式消费上一个异步任务的结果 ###

/**
         * thenAcceptAsync 以异步的方式消费上一个异步任务的结果
         */
        static void thenAcceptAsync() throws ExecutionException, InterruptedException {
        
            final ExecutorService executor = Executors.newFixedThreadPool(3);
            CompletableFuture.supplyAsync(() -> {
        
                System.out.println(Thread.currentThread().getName() + " supplyAsync");
                return "Java";
            }, executor).thenAcceptAsync(e -> {
        
                System.out.println(Thread.currentThread().getName() + " thenAcceptAsync");
                System.out.println(Thread.currentThread().getName() + " " + e);
            });
            executor.shutdown();
        }

> 在任务链的末端，如果执行的任务既不想对上一个任务的输出做进一步处理，又不想消费上一个任务的输出结果。那么可以使用thenRun或者thenRunSync方法来执行Runnable任务。

### thenRun 以异步的而方式执行Runable任务 ###

static void thenRun() throws ExecutionException, InterruptedException {
        
        final ExecutorService executor = Executors.newFixedThreadPool(3);
        CompletableFuture.supplyAsync(() -> {
        
            System.out.println(Thread.currentThread().getName() + " supplyAsync");
            return "Java";
        }, executor).thenAcceptAsync(e -> {
        
            System.out.println(Thread.currentThread().getName() + " thenAcceptAsync");
            System.out.println(Thread.currentThread().getName() + " " + e);
        }).thenRunAsync(() -> {
        
            System.out.println("all of task completed." + Thread.currentThread().getName());
        });
    }

## 合并多个Future ##

CompletableFuture还允许将若干个Future合并成一个Future的使用方式，可以通过thenCompose方法或者thenCombine方法来实现多个Future的合并

### thenCompose 方法示例 ###

void thenCompose() {
        
        //通过thenCompose将两个Future合并成一个Future
        final CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> "Java")
                //s 为上一个Future的计算结果
                .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " scala"));
        //合并后的Future通过thenApply方法组成任务链
        completableFuture.thenApply(String::toUpperCase)
                .thenAccept(System.out::println);
    }

### thenCombine 方法示例 ###

void thenCombine() {
        
        final CompletableFuture<String> thenCombine = CompletableFuture.supplyAsync(() -> "Java")
                .thenCombine(CompletableFuture.supplyAsync(() -> " scala"),
                        //s1为第一个Future计算的结果，s2为第二个Future计算的结果 
                        (s1, s2) -> s1 + s2);
        thenCombine.thenApply(String::toUpperCase)
                .thenAccept(System.out::println);
    }

## 多Future的并行计算 ##

如果想多个独立CompletableFuture同时并行计算，可以借助allOf来完成，其类似于ExecutorService的invokeAll批量提交异步任务。

/**
     * 多Future的并行计算
     */
    void allOf() throws ExecutionException, InterruptedException {
        
        //定义三个CompletableFuture
        final CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "Java");
        final CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "Parallel");
        final CompletableFuture<String> f3 = CompletableFuture.supplyAsync(() -> "Future");
        //批量执行 ，返回值是一个 void 类型的CompletableFuture
        final CompletableFuture<Void> future = CompletableFuture.allOf(f1, f2, f3).thenRun(() -> {
        
            try {
        
                System.out.println(f1.isDone() + " and result " + f1.get());
                System.out.println(f2.isDone() + " and result " + f2.get());
                System.out.println(f3.isDone() + " and result " + f3.get());
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        });
        //阻塞等待运行结束
        future.get();
    }

## CompletableFuture总结 ##

自JDK1.8以来，CompletableFuture的引入不仅很好地填充了Future的不足之处，还提供了非常遍历的异步编程方式，借助于CompletableFuture，我们可以很轻易地开发出异步运行的代码，甚至不用关心地城线程的维护和管理，只需要关注代码函数本身即可。