经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Java » 查看文章
Java中「Future」接口详解
来源：cnblogs　　作者：知了一笑　　时间：2023/4/17 9:16:41　　对本文有异议

目录

主打一手结果导向；

一、背景

在系统中，异步执行任务，是很常见的功能逻辑，但是在不同的场景中，又存在很多细节差异；

有的任务只强调「执行过程」，并不需要追溯任务自身的「执行结果」，这里并不是指对系统和业务产生的效果，比如定时任务、消息队列等场景；

但是有些任务即强调「执行过程」，又需要追溯任务自身的「执行结果」，在流程中依赖某个异步结果，判断流程是否中断，比如「并行」处理；

【串行处理】整个流程按照逻辑逐步推进，如果出现异常会导致流程中断；

【并行处理】主流程按照逻辑逐步推进，其他「异步」交互的流程执行完毕后，将结果返回到主流程，如果「异步」流程异常，会影响部分结果；

此前在《「订单」业务》的内容中，聊过关于「串行」和「并行」的应用对比，即在订单详情的加载过程中，通过「并行」的方式读取：商品、商户、订单、用户等信息，提升接口的响应时间；

二、Future接口

1、入门案例

异步是对流程的解耦，但是有的流程中又依赖异步执行的最终结果，此时就可以使用「Future」接口来达到该目的，先来看一个简单的入门案例；

  1. public class ServerTask implements Callable<Integer> {
  2.     @Override
  3.     public Integer call() throws Exception {
  4.         Thread.sleep(2000);
  5.         return 3;
  6.     }
  7. }
  8. public class FutureBase01 {
  9.     public static void main(String[] args) throws Exception {
  10.         TimeInterval timer = DateUtil.timer();
  11.         // 线程池
  12.         ExecutorService executor = Executors.newFixedThreadPool(3);
  13.         // 批量任务
  14.         List<ServerTask> serverTasks = new ArrayList<>() ;
  15.         for (int i=0;i<3;i++){
  16.             serverTasks.add(new ServerTask());
  17.         }
  18.         List<Future<Integer>> taskResList = executor.invokeAll(serverTasks) ;
  19.         // 结果输出
  20.         for (Future<Integer> intFuture:taskResList){
  21.             System.out.println(intFuture.get());
  22.         }
  23.         // 耗时统计
  24.         System.out.println("timer...interval = "+timer.interval());
  25.     }
  26. }

这里模拟一个场景，以线程池批量执行异步任务，在任务内线程休眠2秒，以并行的方式最终获取全部结果，只耗时2秒多一点，如果串行的话耗时肯定超过6秒；

2、Future接口

Future表示异步计算的结果，提供了用于检查计算是否完成、等待计算完成、以及检索计算结果的方法。

【核心方法】

  • get()：等待任务完成，获取执行结果，如果任务取消会抛出异常；
  • get(long timeout, TimeUnit unit)：指定等待任务完成的时间，等待超时会抛出异常；
  • isDone()：判断任务是否完成；
  • isCancelled()：判断任务是否被取消；
  • cancel(boolean mayInterruptIfRunning)：尝试取消此任务的执行，如果任务已经完成、已经取消或由于其他原因无法取消，则此尝试将失败；

【基础用法】

  1. public class FutureBase02 {
  2.     public static void main(String[] args) throws Exception {
  3.         // 线程池执行任务
  4.         ExecutorService executor = Executors.newFixedThreadPool(3);
  5.         FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
  6.             @Override
  7.             public String call() throws Exception {
  8.                 Thread.sleep(3000);
  9.                 return "task...OK";
  10.             }
  11.         }) ;
  12.         executor.execute(futureTask);
  13.         // 任务信息获取
  14.         System.out.println("是否完成："+futureTask.isDone());
  15.         System.out.println("是否取消："+futureTask.isCancelled());
  16.         System.out.println("获取结果："+futureTask.get());
  17.         System.out.println("尝试取消："+futureTask.cancel(Boolean.TRUE));
  18.     }
  19. }

【FutureTask】

Future接口的基本实现类，提供了计算的启动和取消、查询计算是否完成以及检索计算结果的方法；

在「FutureTask」类中，可以看到线程异步执行任务时，其中的核心状态转换，以及最终结果写出的方式；

虽然「Future」从设计上，实现了异步计算的结果获取，但是通过上面的案例也可以发现，流程的主线程在执行get()方法时会阻塞，直到最终获取结果，显然对于程序来说并不友好；

在JDK1.8提供「CompletableFuture」类，对「Future」进行优化和扩展；

三、CompletableFuture类

1、基础说明

「CompletableFuture」类提供函数编程的能力，可以通过回调的方式处理计算结果，并且支持组合操作，提供很多方法来实现异步编排，降低异步编程的复杂度；

「CompletableFuture」实现「Future」和「CompletionStage」两个接口；

  • Future：表示异步计算的结果；
  • CompletionStage：表示异步计算的一个步骤，当一个阶段计算完成时，可能会触发其他阶段，即步骤可能由其他CompletionStage触发；

【入门案例】

  1. public class CompletableBase01 {
  2.     public static void main(String[] args) throws Exception {
  3.         // 线程池
  4.         ExecutorService executor = Executors.newFixedThreadPool(3);
  5.         // 任务执行
  6.         CompletableFuture<String> cft = CompletableFuture.supplyAsync(() -> {
  7.             try {
  8.                 Thread.sleep(3000);
  9.             } catch (InterruptedException e) {
  10.                 e.printStackTrace();
  11.             }
  12.             return "Res...OK";
  13.         }, executor);
  14.         // 结果输出
  15.         System.out.println(cft.get());
  16.     }
  17. }

2、核心方法

2.1 实例方法

  1. public class Completable01 {
  2.     public static void main(String[] args) throws Exception {
  3.         // 线程池
  4.         ExecutorService executor = Executors.newFixedThreadPool(3);
  6.         // 1、创建未完成的CompletableFuture，通过complete()方法完成
  7.         CompletableFuture<Integer> cft01 = new CompletableFuture<>() ;
  8.         cft01.complete(99) ;
  10.         // 2、创建已经完成CompletableFuture，并且给定结果
  11.         CompletableFuture<String> cft02 = CompletableFuture.completedFuture("given...value");
  13.         // 3、有返回值，默认ForkJoinPool线程池
  14.         CompletableFuture<String> cft03 = CompletableFuture.supplyAsync(() -> {return "OK-3";});
  16.         // 4、有返回值，采用Executor自定义线程池
  17.         CompletableFuture<String> cft04 = CompletableFuture.supplyAsync(() -> {return "OK-4";},executor);
  19.         // 5、无返回值，默认ForkJoinPool线程池
  20.         CompletableFuture<Void> cft05 = CompletableFuture.runAsync(() -> {});
  22.         // 6、无返回值，采用Executor自定义线程池
  23.         CompletableFuture<Void> cft06 = CompletableFuture.runAsync(()-> {}, executor);
  24.     }
  25. }

2.2 计算方法

  1. public class Completable02 {
  2.     public static void main(String[] args) throws Exception {
  3.         // 线程池
  4.         ExecutorService executor = Executors.newFixedThreadPool(3);
  5.         CompletableFuture<String> cft01 = CompletableFuture.supplyAsync(() -> {
  6.             try {
  7.                 Thread.sleep(2000);
  8.             } catch (InterruptedException e) {
  9.                 e.printStackTrace();
  10.             }
  11.             return "OK";
  12.         },executor);
  14.         // 1、计算完成后，执行后续处理
  15.         // cft01.whenComplete((res, ex) -> System.out.println("Result："+res+"；Exe："+ex));
  17.         // 2、触发计算，如果没有完成，则get设定的值，如果已完成，则get任务返回值
  18.         // boolean completeFlag = cft01.complete("given...value");
  19.         // if (completeFlag){
  20.         //     System.out.println(cft01.get());
  21.         // } else {
  22.         //     System.out.println(cft01.get());
  23.         // }
  25.         // 3、开启新CompletionStage，重新获取线程执行任务
  26.         cft01.whenCompleteAsync((res, ex) -> System.out.println("Result："+res+"；Exe："+ex),executor);
  27.     }
  28. }

2.3 结果获取方法

  1. public class Completable03 {
  2.     public static void main(String[] args) throws Exception {
  3.         // 线程池
  4.         ExecutorService executor = Executors.newFixedThreadPool(3);
  5.         CompletableFuture<String> cft01 = CompletableFuture.supplyAsync(() -> {
  6.             try {
  7.                 Thread.sleep(2000);
  8.             } catch (InterruptedException e) {
  9.                 e.printStackTrace();
  10.             }
  11.             return "Res...OK";
  12.         },executor);
  13.         // 1、阻塞直到获取结果
  14.         // System.out.println(cft01.get());
  16.         // 2、设定超时的阻塞获取结果
  17.         // System.out.println(cft01.get(4, TimeUnit.SECONDS));
  19.         // 3、非阻塞获取结果，如果任务已经完成，则返回结果，如果任务未完成，返回给定的值
  20.         // System.out.println(cft01.getNow("given...value"));
  22.         // 4、get获取抛检查异常，join获取非检查异常
  23.         System.out.println(cft01.join());
  24.     }
  25. }

2.4 任务编排方法

  1. public class Completable04 {
  2.     public static void main(String[] args) throws Exception {
  3.         // 线程池
  4.         ExecutorService executor = Executors.newFixedThreadPool(3);
  5.         CompletableFuture<String> cft01 = CompletableFuture.supplyAsync(() -> {
  6.             try {
  7.                 Thread.sleep(2000);
  8.             } catch (InterruptedException e) {
  9.                 e.printStackTrace();
  10.             }
  11.             System.out.println("OK-1");
  12.             return "OK";
  13.         },executor);
  15.         // 1、cft01任务执行完成后，执行之后的任务，此处不关注cft01的结果
  16.         // cft01.thenRun(() -> System.out.println("task...run")) ;
  18.         // 2、cft01任务执行完成后，执行之后的任务，可以获取cft01的结果
  19.         // cft01.thenAccept((res) -> {
  20.         //     System.out.println("cft01："+res);
  21.         //     System.out.println("task...run");
  22.         // });
  24.         // 3、cft01任务执行完成后，执行之后的任务，获取cft01的结果，并且具有返回值
  25.         // CompletableFuture<Integer> cft02 = cft01.thenApply((res) -> {
  26.         //     System.out.println("cft01："+res);
  27.         //     return 99 ;
  28.         // });
  29.         // System.out.println(cft02.get());
  31.         // 4、顺序执行cft01、cft02
  32.         // CompletableFuture<String> cft02 = cft01.thenCompose((res) ->  CompletableFuture.supplyAsync(() -> {
  33.         //     System.out.println("cft01："+res);
  34.         //     return "OK-2";
  35.         // }));
  36.         // cft02.whenComplete((res,ex) -> System.out.println("Result："+res+"；Exe："+ex));
  38.         // 5、对比任务的执行效率，由于cft02先完成，所以取cft02的结果
  39.         // CompletableFuture<String> cft02 = cft01.applyToEither(CompletableFuture.supplyAsync(() -> {
  40.         //     System.out.println("run...cft02");
  41.         //     try {
  42.         //         Thread.sleep(3000);
  43.         //     } catch (InterruptedException e) {
  44.         //         e.printStackTrace();
  45.         //     }
  46.         //     return "OK-2";
  47.         // }),(res) -> {
  48.         //     System.out.println("either...result：" + res);
  49.         //     return res;
  50.         // });
  51.         // System.out.println("finally...result：" + cft02.get());
  53.         // 6、两组任务执行完成后，对结果进行合并
  54.         // CompletableFuture<String> cft02 = CompletableFuture.supplyAsync(() -> "OK-2") ;
  55.         // String finallyRes = cft01.thenCombine(cft02,(res1,res2) -> {
  56.         //     System.out.println("res1："+res1+"；res2："+res2);
  57.         //     return res1+"；"+res2 ;
  58.         // }).get();
  59.         // System.out.println(finallyRes);
  62.         CompletableFuture<String> cft02 = CompletableFuture.supplyAsync(() -> {
  63.             System.out.println("OK-2");
  64.             return  "OK-2";
  65.         }) ;
  66.         CompletableFuture<String> cft03 = CompletableFuture.supplyAsync(() -> {
  67.             System.out.println("OK-3");
  68.             return "OK-3";
  69.         }) ;
  70.         // 7、等待批量任务执行完返回
  71.         // CompletableFuture.allOf(cft01,cft02,cft03).get();
  73.         // 8、任意一个任务执行完即返回
  74.         System.out.println("Sign："+CompletableFuture.anyOf(cft01,cft02,cft03).get());
  75.     }
  76. }

2.5 异常处理方法

  1. public class Completable05 {
  2.     public static void main(String[] args) throws Exception {
  3.         // 线程池
  4.         ExecutorService executor = Executors.newFixedThreadPool(3);
  5.         CompletableFuture<String> cft01 = CompletableFuture.supplyAsync(() -> {
  6.             if (1 > 0){
  7.                 throw new RuntimeException("task...exception");
  8.             }
  9.             return "OK";
  10.         },executor);
  12.         // 1、捕获cft01的异常信息，并提供返回值
  13.         String finallyRes = cft01.thenApply((res) -> {
  14.             System.out.println("cft01-res：" + res);
  15.             return res;
  16.         }).exceptionally((ex) -> {
  17.             System.out.println("cft01-exe：" + ex.getMessage());
  18.             return "error" ;
  19.         }).get();
  20.         System.out.println("finallyRes="+finallyRes);
  23.         CompletableFuture<String> cft02 = CompletableFuture.supplyAsync(() -> {
  24.             try {
  25.                 Thread.sleep(1000);
  26.             } catch (InterruptedException e) {
  27.                 e.printStackTrace();
  28.             }
  29.             return "OK-2";
  30.         },executor);
  31.         // 2、如果cft02未完成，则get时抛出指定异常信息
  32.         boolean exeFlag = cft02.completeExceptionally(new RuntimeException("given...exception"));
  33.         if (exeFlag){
  34.             System.out.println(cft02.get());
  35.         } else {
  36.             System.out.println(cft02.get());
  37.         }
  38.     }
  39. }

3、线程池问题

  • 在实践中，通常不使用ForkJoinPool#commonPool()公共线程池，会出现线程竞争问题，从而形成系统瓶颈；
  • 在任务编排中，如果出现依赖情况或者父子任务，尽量使用多个线程池，从而避免任务请求同一个线程池，规避死锁情况发生；

四、CompletableFuture原理

1、核心结构

在分析「CompletableFuture」其原理之前，首先看一下涉及的核心结构；

【CompletableFuture】

在该类中有两个关键的字段：「result」存储当前CF的结果，「stack」代表栈顶元素，即当前CF计算完成后会触发的依赖动作；从上面案例中可知，依赖动作可以没有或者有多个；

【Completion】

依赖动作的封装类；

【UniCompletion】

继承Completion类，一元依赖的基础类，「executor」指线程池，「dep」指依赖的计算，「src」指源动作；

【BiCompletion】

继承UniCompletion类，二元或者多元依赖的基础类，「snd」指第二个源动作；

2、零依赖

顾名思义，即各个CF之间不产生依赖关系；

  1. public class DepZero {
  2.     public static void main(String[] args) throws Exception {
  3.         ExecutorService executor = Executors.newFixedThreadPool(3);
  4.         CompletableFuture<String> cft1 = CompletableFuture.supplyAsync(()-> "OK-1",executor);
  5.         CompletableFuture<String> cft2 = CompletableFuture.supplyAsync(()-> "OK-2",executor);
  6.         System.out.println(cft1.get()+";"+cft2.get());
  7.     }
  8. }

3、一元依赖

即CF之间的单个依赖关系；这里使用「thenApply」方法演示，为了看到效果，使「cft1」长时间休眠，断点查看「stack」结构；

  1. public class DepOne {
  2.     public static void main(String[] args) throws Exception {
  3.         ExecutorService executor = Executors.newFixedThreadPool(3);
  4.         CompletableFuture<String> cft1 = CompletableFuture.supplyAsync(() -> {
  5.             try {
  6.                 Thread.sleep(30000);
  7.             } catch (InterruptedException e) {
  8.                 e.printStackTrace();
  9.             }
  10.             return "OK-1";
  11.         },executor);
  13.         CompletableFuture<String> cft2 = cft1.thenApply(res -> {
  14.             System.out.println("cft01-res"+res);
  15.             return "OK-2" ;
  16.         });
  17.         System.out.println("cft02-res"+cft2.get());
  18.     }
  19. }

断点截图：

原理分析：

观察者Completion注册到「cft1」，注册时会检查计算是否完成，未完成则观察者入栈，当「cft1」计算完成会弹栈；已完成则直接触发观察者；

可以调整断点代码，让「cft1」先处于完成状态，再查看其运行时结构，从而分析完整的逻辑；

4、二元依赖

即一个CF同时依赖两个CF；这里使用「thenCombine」方法演示；为了看到效果，使「cft1、cft2」长时间休眠，断点查看「stack」结构；

  1. public class DepTwo {
  2.     public static void main(String[] args) throws Exception {
  3.         ExecutorService executor = Executors.newFixedThreadPool(3);
  4.         CompletableFuture<String> cft1 = CompletableFuture.supplyAsync(() -> {
  5.             try {
  6.                 Thread.sleep(30000);
  7.             } catch (InterruptedException e) {
  8.                 e.printStackTrace();
  9.             }
  10.             return "OK-1";
  11.         },executor);
  12.         CompletableFuture<String> cft2 = CompletableFuture.supplyAsync(() -> {
  13.             try {
  14.                 Thread.sleep(30000);
  15.             } catch (InterruptedException e) {
  16.                 e.printStackTrace();
  17.             }
  18.             return "OK-2";
  19.         },executor);
  21.         // cft3 依赖 cft1和cft2 的计算结果
  22.         CompletableFuture<String> cft3 = cft1.thenCombine(cft2,(res1,res2) -> {
  23.             System.out.println("cft01-res:"+res1);
  24.             System.out.println("cft02-res:"+res2);
  25.             return "OK-3" ;
  26.         });
  27.         System.out.println("cft03-res:"+cft3.get());
  28.     }
  29. }

断点截图：

原理分析：

在「cft1」和「cft2」未完成的状态下，尝试将BiApply压入「cft1」和「cft2」两个栈中，任意CF完成时，会尝试触发观察者，观察者检查「cft1」和「cft2」是否都完成，如果完成则执行；

5、多元依赖

即一个CF同时依赖多个CF；这里使用「allOf」方法演示；为了看到效果，使「cft1、cft2、cft3」长时间休眠，断点查看「stack」结构；

  1. public class DepMore {
  2.     public static void main(String[] args) throws Exception {
  3.         ExecutorService executor = Executors.newFixedThreadPool(3);
  4.         CompletableFuture<String> cft1 = CompletableFuture.supplyAsync(() -> {
  5.             try {
  6.                 Thread.sleep(30000);
  7.             } catch (InterruptedException e) {
  8.                 e.printStackTrace();
  9.             }
  10.             return "OK-1";
  11.         },executor);
  12.         CompletableFuture<String> cft2 = CompletableFuture.supplyAsync(() -> {
  13.             try {
  14.                 Thread.sleep(30000);
  15.             } catch (InterruptedException e) {
  16.                 e.printStackTrace();
  17.             }
  18.             return "OK-2";
  19.         },executor);
  21.         CompletableFuture<String> cft3 = CompletableFuture.supplyAsync(() -> {
  22.             try {
  23.                 Thread.sleep(30000);
  24.             } catch (InterruptedException e) {
  25.                 e.printStackTrace();
  26.             }
  27.             return "OK-3";
  28.         },executor);
  30.         // cft4 依赖 cft1和cft2和cft3 的计算结果
  31.         CompletableFuture<Void> cft4 = CompletableFuture.allOf(cft1,cft2,cft3);
  32.         CompletableFuture<String> finallyRes = cft4.thenApply(tm -> {
  33.             System.out.println("cft01-res:"+cft1.join());
  34.             System.out.println("cft02-res:"+cft2.join());
  35.             System.out.println("cft03-res:"+cft3.join());
  36.             return "OK-4";
  37.         });
  38.         System.out.println("finally-res:"+finallyRes.get());
  39.     }
  40. }

断点截图：

原理分析：

多元依赖的回调方法除了「allOf」还有「anyOf」，其实现原理都是将依赖的多个CF补全为平衡二叉树，从断点图可知会按照树的层级处理，核心结构参考二元依赖即可；

五、参考源码

  1. 编程文档：
  2. https://gitee.com/cicadasmile/butte-java-note
  4. 应用仓库：
  5. https://gitee.com/cicadasmile/butte-flyer-parent

原文链接：https://www.cnblogs.com/cicada-smile/p/17324695.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Java热门文章

  • 【IDEA】Application context not configured for this file
  • [IDEA] chapter_reader - idea看小说插件 idea阅读插件 idea摸鱼插件
  • thrift常见异常及原因分析
  • Java 调用 PaddleDetection 模型
  • JVM 详解，大白话带你认识 JVM
  • spring-boot-2.0.3不一样系列之番外篇 - @Configuration、Condition与@Conditional
  • Java Comparator.comparing比较导致空指针异常的解决
  • Java使用EasyExcel进行单元格合并的问题详解
  • 死磕 java集合之ArrayBlockingQueue源码分析
  • Java基础之集合Set详解

    • Java推荐文章

  • 面试场景题：一次关于线程池使用场景的讨论。
  • 前端黑科技：使用 JavaScript 实现网页扫码功能
  • 聊一聊 Netty 数据搬运工 ByteBuf 体系的设计与实现
  • 简单设计一个JAVA并行处理工具类
  • 【杂谈】JPA乐观锁改悲观锁遇到的一些问题与思考
  • 认识netty的基本组件
  • Jenkins如何使用CrumbIssuer防御CSRF攻击
  • 如何平稳地从nacos迁移到r-nacos？
  • 线程池遇到父子任务，有大坑，要注意！
  • Java JVM——12. 垃圾回收理论概述

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号