CompletableFuture使用详解

柔情只为你懂 2023-10-03 13:11 10阅读 0赞

[1、CompletableFuture 使用详解][1_CompletableFuture]  
[2、Java8新的异步编程方式 CompletableFuture(一)][2_Java8_ CompletableFuture]  
[3、CompletableFuture使用详解][3_CompletableFuture]

### CompletableFuture 使用详解 ###

*  一 简介
 *  二 异步操作
 *  三 异步操作案例
 *   *  1.runAsync方法不支持返回值
     *   *  案例代码
         *  结果分析
     *  2.supplyAsync方法支持返回值
     *   *  案例代码
         *  结果分析
 *  三 CompletableFuture常用方法
 *   *  3.1.依赖关系
     *   *  3.1.1.thenApply
         *  3.1.2.thenCompose
     *  3.2.and集合关系
     *  3.3.or聚合关系
     *  3.4.并行执行
     *  3.5.结果处理
 *  四 案例源码

# 一 简介 #

`CompletableFuture`是对Future的扩展和增强。`CompletableFuture`实现了`Future`接口，并在此基础上进行了丰富的扩展，完美弥补了`Future`的局限性，同时`CompletableFuture`实现了对任务编排的能力。执行某一阶段，可以向下执行后续阶段。异步执行的，默认线程池是`ForkJoinPool.commonPool()`，但为了业务之间互不影响，且便于定位问题，强烈推荐使用自定义线程池。

实际项目中我们在对接口优化时，我们可能会将一些串行的操作进行并行处理，提高效率。  
比如：

> 1、查询数据字典，查询组织对应需要转成map，这两个方法之间无关联就可以做成并行处理。  
> 2、在进行插入数据库时，一个主表两个关联表都需要插入数据，我们处理好数据之前的关联关系，在入 库时也可以做成并行操作。

# 二 异步操作 #

CompletableFuture的静态工厂方法

<table> 
 <thead> 
  <tr> 
   <th align="left">方法名</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">CompletableFuture&lt; Void&gt; <code>runAsync</code>(Runnable runnable)</td> 
   <td align="left">使用<code>ForkJoinPool.commonPool()</code>作为它的线程池执行异步代码。</td> 
  </tr> 
  <tr> 
   <td align="left">CompletableFuture&lt; Void&gt; <code>runAsync</code>(Runnable runnable, Executor executor)</td> 
   <td align="left">使用指定的<code>thread pool</code>执行异步代码。</td> 
  </tr> 
  <tr> 
   <td align="left">CompletableFuture&lt; U&gt; <code>supplyAsync</code>(Supplier&lt; U&gt; supplier)</td> 
   <td align="left">使用<code>ForkJoinPool.commonPool()</code>作为它的线程池执行异步代码，异步操作有返回值</td> 
  </tr> 
  <tr> 
   <td align="left">CompletableFuture&lt; U&gt; <code>supplyAsync</code>(Supplier&lt; U&gt; supplier, Executor executor)</td> 
   <td align="left">使用指定的<code>thread pool</code>执行异步代码，异步操作有返回值</td> 
  </tr> 
 </tbody> 
</table>

四个方法的区别：

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

# 三 异步操作案例 #

案例中使用了[自定的线程池][Link 1]

## 1.runAsync方法不支持返回值 ##

### 案例代码 ###

//使用自定义的线程池
        @Autowired
        private ThreadPoolExecutor executor;
        @Test
        void runAsyncTest() throws Exception {
        
            //runAsync方法不支持返回值
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
        
                //{}代码块就是一个方法，对应我们实际项目中可以是抽取的方法
                System.out.println("0---runAsync--方法一开始执行了");
                try {
        
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
        
                }
                System.out.println("①---runAsync--方法一执行结束了");
            }, executor);
            CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
        
                System.out.println("---runAsync--方法四开始执行了");
                try {
        
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
        
                }
                System.out.println("---runAsync--方法四执行结束了");
            }, executor);
            System.out.println("②---runAsync--方法二执行了");
            System.out.println(future.get());
            future2.get();
            System.out.println("③---runAsync--方法三执行了");
        }

### 结果分析 ###

![在这里插入图片描述][421aa7e1f1084da8ab803116925fec2c.png]

## 2.supplyAsync方法支持返回值 ##

### 案例代码 ###

@Test
        void supplyAsyncTest() throws Exception {
        
            //方法一
            CompletableFuture<Long> future1 = CompletableFuture.supplyAsync(() -> {
        
                try {
        
                    System.out.println("①---supplyAsync--方法一开始执行");
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
        
                }
                System.out.println("①---supplyAsync--方法一执行结束");
                //实际项目中将方法执行的结果返回,有且只能返回一个结果
                return System.currentTimeMillis();
            },executor);
            //方法二
            CompletableFuture<Long> future2 = CompletableFuture.supplyAsync(() -> {
        
                try {
        
                    System.out.println("②---supplyAsync--方法二开始执行");
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
        
                }
                System.out.println("②---supplyAsync--方法二执行结束");
                return System.currentTimeMillis();
            },executor);
            //方法三
            System.out.println("③---supplyAsync--方法三执行了");
            System.out.println("future1的supplyAsync方法一执行结束了 = " + future1.get());
            System.out.println("future2的supplyAsync方法二执行结束了 = " + future2.get());
            //上述的future会阻塞，只有上面执行完了，才会执行方法四
            System.out.println("④---supplyAsync--方法四执行了");
        }

### 结果分析 ###

![结果分析][73191ffea3664913a55eb0f44da6f9aa.png]

# 三 CompletableFuture常用方法 #

## 3.1.依赖关系 ##

*  `thenApply()`：把前面任务的执行结果，交给后面的`Function`
 *  `thenCompose()`：用来连接两个有依赖关系的任务，结果由第二个任务返回

这两个都属于**结果转换**将上一段任务的执行结果作为下一阶段任务的入参参与重新计算，产生新的结果。

### 3.1.1.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)

`thenApply`和`thenApplyAsync`区别在于，使用`thenApply`方法时子任务与父任务使用的是同一个线程，而`thenApplyAsync`在子任务中是另起一个线程执行任务，并且`thenApplyAsync`可以自定义线程池，默认的使用`ForkJoinPool.commonPool()`线程池。  
**案例代码**

@Test
        void thenApplyTest() throws Exception {
        
            CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        
                int result = 100;
                System.out.println(Thread.currentThread().getName() + "：future第一次运算：" + result);
                return result;
            }, executor).thenApply(number -> {
        
                //使用`thenApply`方法时子任务与父任务使用的是同一个线程
                int result = number * 3;
                System.out.println(Thread.currentThread().getName() + "：future第二次运算：" + result);
                return result;
            });
    
            CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        
                int result = 100;
                System.out.println(Thread.currentThread().getName() + "：future2第一次运算：" + result);
                return result;
            }, executor).thenApplyAsync(number -> {
        
                //`thenApplyAsync`在子任务中是另起一个线程执行任务,并且`thenApplyAsync`可以自定义线程池
                int result = number * 3;
                System.out.println(Thread.currentThread().getName() + "：future2第二次运算：" + result);
                return result;
            }, executor);
    
            System.out.println("future结果：" + future.get());
            System.out.println("future2结果：" + future2.get());
            System.out.println("任务3执行了 = ");
        }

**结果分析**  
![在这里插入图片描述][5a88f9271b274e0ba0a2874ab8e3473a.png]

### 3.1.2.thenCompose ###

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

**常用方法：**

public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn);
    public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStag

**案例代码**

@Test
        void thenComposeTest() throws Exception {
        
            CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
        
                @Override
                public Integer get() {
        
                    int number = new Random().nextInt(30);
                    System.out.println(Thread.currentThread().getName() + "：第一次运算：" + number);
                    return number;
                }
            }, executor);
            CompletableFuture<Integer> future2 = future1.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(Thread.currentThread().getName() + "：第二次运算：" + number);
                            return number;
                        }
                    }, executor);
                }
            });
            System.out.println("future1.get() = " + future1.get());
            System.out.println("future2.get() = " + future2.get());
        }

**结果分析**

![在这里插入图片描述][9393e95babfd4a5783bdc527add05baa.png]

## 3.2.and集合关系 ##

*  `thenCombine()`：合并任务，有返回值
 *  `thenAccepetBoth()`：两个任务执行完成后，将结果交给`thenAccepetBoth`处理，无返回值
 *  `runAfterBoth()`：两个任务都执行完成后，执行下一步操作(`Runnable`类型任务)

## 3.3.or聚合关系 ##

*  `applyToEither()`：两个任务哪个执行的快，就使用哪一个结果，有返回值
 *  `acceptEither()`：两个任务哪个执行的快，就消费哪一个结果，无返回值
 *  `runAfterEither()`：任意一个任务执行完成，进行下一步操作(`Runnable`类型任务)

## 3.4.并行执行 ##

*  `allOf()`：当所有给定的 `CompletableFuture` 完成时，返回一个新的 `CompletableFuture`
 *  `anyOf()`：当任何一个给定的`CompletablFuture`完成时，返回一个新的`CompletableFuture`

## 3.5.结果处理 ##

*  `whenComplete`：当任务完成时，将使用结果(或 null)和此阶段的异常(或 null如果没有)执行给定操作
 *  `exceptionally`：返回一个新的`CompletableFuture`，当前面的`CompletableFuture`完成时，它也完成，当它异常完成时，给定函数的异常触发这个`CompletableFuture`的完成

# 四 案例源码 #

[项目案例源码地址][Link 2]

[1_CompletableFuture]: https://www.jianshu.com/p/6bac52527ca4
[2_Java8_ CompletableFuture]: https://www.jianshu.com/p/dff9063e1ab6
[3_CompletableFuture]: https://blog.csdn.net/sermonlizhi/article/details/123356877
[Link 1]: https://blog.csdn.net/weixin_43811057/article/details/129580537?csdn_share_tail=%7B%22type%22:%22blog%22,%22rType%22:%22article%22,%22rId%22:%22129580537%22,%22source%22:%22weixin_43811057%22%7D
[421aa7e1f1084da8ab803116925fec2c.png]: https://img-blog.csdnimg.cn/421aa7e1f1084da8ab803116925fec2c.png
[73191ffea3664913a55eb0f44da6f9aa.png]: https://img-blog.csdnimg.cn/73191ffea3664913a55eb0f44da6f9aa.png
[5a88f9271b274e0ba0a2874ab8e3473a.png]: https://img-blog.csdnimg.cn/5a88f9271b274e0ba0a2874ab8e3473a.png
[9393e95babfd4a5783bdc527add05baa.png]: https://img-blog.csdnimg.cn/9393e95babfd4a5783bdc527add05baa.png
[Link 2]: https://gitee.com/wl8888/completable-future