JUC并发编程之CompletableFuture

Love The Way You Lie 2023-09-30 23:33 32阅读 0赞

> ### Future ###
> 
> future是java5新加的一个接口，他提供了一种异步并行计算的功能
> 
> 接口定义了操作异步任务执行的一些方法，如获取异步任务的执行结果、取消任务的执行、判断任务是否被取消、判断任务是否执行完毕
> 
> 目的：异步多线程执行且有返回结果，特点：多线程/有返回/异步任务
> 
> 补充：Runnable实现的是run方法，没有返回值，没有异常，Callable实现的是call方法，有返回值，需要处理异常
> 
> Future接口常用实现类Future Task异步任务![67ab60af236f43868b83c95947f3faa0.png][]
> 
> 代码实现
> 
>     public class CompletableFutureDemo {
>         public static void main(String[] args) throws ExecutionException, InterruptedException {
>             FutureTask<String> futureTask = new FutureTask<>(new MyThread());
>             Thread t1 = new Thread(futureTask,"t1");
>             t1.start();
>             System.out.println(futureTask.get());
>         }
>     }
>     
>     
>     
>     class MyThread implements Callable<String>{
>     
>         @Override
>         public String call() throws Exception {
>             System.out.println("----come in call()");
>             return "hello Callable";
>         }
>     }
> 
> future的优缺点
> 
> NO: 1、get方法容易阻塞：一旦调用，如果计算没有完成，容易程序阻塞
> 
> 2、isDone轮询：轮询的方式会耗费CPU资源，而且也不见得能及时得到计算结果
> 
> 小结：future对于结果的获取不是很友好，只能通过阻塞或轮询的方式得到任务结果
> 
> 注：实际程序中轮询的方式并不比阻塞好，这么写是给用户看的，用户不可能看你的程序就停在那了，这么写可以给个提示，也不会直接异常

> ### CompletableFuture的异步优化思路 ###
> 
> 背景：对于简单的业务场景使用Future完全OK，但是对于复杂的业务场景，比如：
> 
> 1、回调通知：应对Future的完成时间，完成了可以告诉我，也就是我们的回调通知，通过轮询的方式去判断任务是否完成这样非常的占用CPU并且代码也不优雅
> 
> 2、创建异步任务：Future+线程池配合
> 
> 3、多个任务前后依赖可以组合处理：想将多个异步任务的计算结果组合起来，后一个异步任务的计算结果需要前一个异步任务结果的值，将两个或多个异步计算合成一个异步计算，这几个异步计算相互独立，同时后面这个又依赖前一个处理的结果
> 
> 4、对计算速度最快：当Future集合中某个任务最快结束时，返回结果，返回第一名处理结果
> 
> CompletableFuture提供了一种观察者模式类似的机制，可以让任务完成后通知监听的一方
> 
> 类架构说明
> 
> ![cbb2baed690d48ffb218126b9c019382.png][]
> 
> 接口CompletionStage
> 
> 1、CompletionStage代表异步计算过程中某一个阶段，一个阶段完成以后可能会触发另外一个阶段
> 
> 2、一个阶段的执行可以是一个Function，Consumer或者Runnable
> 
> 3、一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发
> 
> 代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段
> 
> 类CompletableFuture
> 
> 1、在java8当中，CompletableFuture提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合ComplttableFuture的方法
> 
> 2、他可能代表一个明确完成的Future，也有可能代表一个完成阶段，它支持在计算完成以后触发一些函数或执行某个动作
> 
> 3、它实现了Future和CompletionStage接口
> 
> ![b6612e63c25c4f9398b227335e25453e.png][]
> 
> 注：从java8开始引入了CompletableFuture，他是Future的功能增强版，减少阻塞和轮询，可以传入回调对象，当异步任务完成或者发生异常时，自动调用回调对象的回调方法
> 
> 优点：
> 
> 1、异步任务结束时，会自动回调某个对象的方法
> 
> 2、主线程设置好回调后，不再关心异步任务的执行，异步任务之间可以顺序执行
> 
> 3、异步任务出错时，会自动回调某个对象的方法
> 
> 函数式接口复习 ![e190e14b7e874adb8e99dbabc4647477.png][]

### CompletableFuture常用方法 ###

#### 1、获得结果和触发计算 ####

public class CompletableFutureAPIDemo {
        public static void main(String[] args) {
            CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "abc";
            });
    
            //System.out.println(completableFuture.get());   阻塞主线程获取结果
            //System.out.println(completableFuture.get(2L,TimeUnit.SECONDS)); 等待指定时间获取结果
            //System.out.println(completableFuture.join());  阻塞线程获取结果
            System.out.println(completableFuture.getNow("xxx")); //立即获取结果不阻塞，如果没计算完成返回XXX
            System.out.println(completableFuture.complete("completeValue")+"\t"+completableFuture.join());
            //是否打断get方法立刻获得括号值
        }
    }

###  ###

#### 2、对计算结果进行处理 ####

public class CompletableFutureAPI2Demo {
        public static void main(String[] args) {
    
            ExecutorService threadPool = Executors.newFixedThreadPool(3);
    
            //thenApply
            //计算结果存在依赖关系，这两个线程串行化
            //异常相关：由于存在依赖关系（当前步错，不走下一步），当前步骤有异常的话就叫停
            //handle:区别：有异常也可以往下一步走，根据带的异常参数可以进一步处理
            CompletableFuture.supplyAsync(() -> {
                try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}
                System.out.println("111");
                return 1;
            },threadPool).handle((f,e) -> {
                int i = 10/0;
                System.out.println("222");
                return f+2;
            }).thenApply(f -> {
                System.out.println("333");
                return f+3;
            }).whenComplete((v,e) -> {
                if (e == null){
                    System.out.println("------计算结果:"+v);
                }
            }).exceptionally(e -> {
                e.printStackTrace();
                System.out.println("e = " + e.getMessage());
                return null;
            });
            threadPool.shutdown();
            System.out.println(Thread.currentThread().getName()+"-----主线程先去忙其他任务");
        }
    }

#### 3、对计算结果进行消费 ####

//接受任务的处理结果，并消费处理，无返回结果
    public class CompletableFutureAPI3Demo {
        public static void main(String[] args) {
    //        CompletableFuture.supplyAsync(() -> {
    //            return 1;
    //        }).thenApply(f -> {
    //            return f+2;
    //        }).thenApply(f -> {
    //            return f+3;
    //        }).thenAccept(System.out::println);
    
            //任务之间的顺序执行
            /**
             * 1、thenRun：任务A执行完执行任务B，并且B不需要A的结果
             * 2、thenAccept：任务  A执行完执行B,B需要A的结果，但是任务B无返回值
             * 3、thenApply：任务A执行完执行B，B需要A的结果，同时任务B有返回值
             */
    
            System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenRun(() -> {}).join());
            System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenAccept(r-> System.out.println("r = " + r)).join());
            System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenApply(r -> r+"ResuleB").join());
    
        }
    }

#### 4、对计算速度进行选用 ####

public class CompletableFutureFastDemo {
        public static void main(String[] args) {
            CompletableFuture<String> playA = CompletableFuture.supplyAsync(() -> {
                System.out.println("A come in");
                try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}
                return "playA";
            });
    
            CompletableFuture<String> playB = CompletableFuture.supplyAsync(() -> {
                System.out.println("B come in");
                try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}
                return "playB";
            });
            CompletableFuture<String> result = playA.applyToEither(playB, f -> f  + " is winer");
            System.out.println(Thread.currentThread().getName()+"\t"+"--------"+result.join());
        }
    }

#### 5、对计算结果进行合并 ####

public class CompletableFutureCombineDemo {
        public static void main(String[] args) {
            CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
                System.out.println(Thread.currentThread().getName() + "\t ---启动");
                try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}
                return 10;
            });
    
            CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {
                System.out.println(Thread.currentThread().getName() + "\t ---启动");
                try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}
                return 20;
            });
    
            CompletableFuture<Integer> result = completableFuture1.thenCombine(completableFuture2, (x, y) -> {
                System.out.println("开始两个结果合并:" + x + y);
                return x + y;
            });
            System.out.println("result.join() = " + result.join());
        }
    }

注：CompletableFuture线程池运行选择

1、没有传入自定义线程池，都用默认线程池ForkJoinPool

2、传入了一个自定义线程池

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

调用thenRun方法执行第二个任务时，第二个任务和第一个任务使用的是同一个线程池

调用thenRunAsync方法执行第二个任务时，则第一个任务使用的是你自己传入的线程池，第二个任务使用的是ForkJoin线程池

备注：

有可能处理太快，系统优化切换原则，直接使用main线程处理

public class CompletableFutureWithThreadPoolDemo {
        public static void main(String[] args) {
            ExecutorService threadPool = Executors.newFixedThreadPool(5);
    
            try {
                CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> {
                    try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
                    System.out.println("1号任务" + "\t" + Thread.currentThread().getName());
                    return "abcd";
                },threadPool).thenRunAsync(() -> {
                    try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
                    System.out.println("2号任务" + "\t" + Thread.currentThread().getName());
                }).thenRun(() -> {
                    try {TimeUnit.MILLISECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}
                    System.out.println("3号任务" + "\t" + Thread.currentThread().getName());
                }).thenRun(() -> {
                    try {TimeUnit.MILLISECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}
                    System.out.println("4号任务" + "\t" + Thread.currentThread().getName());
                });
                System.out.println(completableFuture.get(2L, TimeUnit.SECONDS));
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                threadPool.shutdown();
            }
        }
    }

[67ab60af236f43868b83c95947f3faa0.png]: https://img-blog.csdnimg.cn/67ab60af236f43868b83c95947f3faa0.png
[cbb2baed690d48ffb218126b9c019382.png]: https://img-blog.csdnimg.cn/cbb2baed690d48ffb218126b9c019382.png
[b6612e63c25c4f9398b227335e25453e.png]: https://img-blog.csdnimg.cn/b6612e63c25c4f9398b227335e25453e.png
[e190e14b7e874adb8e99dbabc4647477.png]: https://img-blog.csdnimg.cn/e190e14b7e874adb8e99dbabc4647477.png