CompletableFuture 异步加载的实现

爱被打了一巴掌 2024-03-26 21:07 87阅读 0赞

本文介绍了CompletableFuture并行开发的基本原理，并提供了一些实践案例，希望对大家理解并行开发提供一些帮助。

### 问题背景 ###

*  APP首页作为前台核心导购流程，需要将 【附近门店】，【推荐商品】，【banner图】，【活动页】等多个中台能力聚合成更具象化的业务流程；
 *  但是各个中台的能力接口性能不尽相同（因为中台不关心业务）：比如附近门店接口99线接近300ms，推荐商品99线接近200ms
 *  基于以上2点，业务研发就面临了一个核心问题：如果将多个性能参差不齐的中台接口聚合成用户体验好，性能高的业务流程

### 核心接口依赖分析 ###

<table> 
 <thead> 
  <tr> 
   <th style="width:92px;">功能</th> 
   <th style="width:210px;">依赖接口</th> 
   <th>平响（ms）</th> 
   <th>99线（ms）</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="width:92px;">附近门店</td> 
   <td style="width:210px;">shop/nearbyStore</td> 
   <td>66</td> 
   <td>101</td> 
   <td>根据经纬度获取附近门店</td> 
  </tr> 
  <tr> 
   <td style="width:92px;">推荐商品</td> 
   <td style="width:210px;">product/getRecommendCommodities</td> 
   <td>170</td> 
   <td>260</td> 
   <td>获取销量前五的商品</td> 
  </tr> 
  <tr> 
   <td style="width:92px;">活动信息</td> 
   <td style="width:210px;">activity/activityInformation</td> 
   <td>62</td> 
   <td>121</td> 
   <td>获取配置的活动</td> 
  </tr> 
  <tr> 
   <td style="width:92px;">banner图</td> 
   <td style="width:210px;">operation/getBanner</td> 
   <td>24</td> 
   <td>43</td> 
   <td>获取APP Banner</td> 
  </tr> 
 </tbody> 
</table>

如大家所见，如果要渲染出所有的信息，则平响至少需要322ms; 99线至少需要：525ms，作为APP核心入口之一，这样的性能显然是不过关的。

### 如何解决性能问题 ###

#### 寻找解决方案 ####

面对该问题时尝试做了以下推演：

1.  敦促下游依赖提升服务性能：
2.  将页面所需数据进行分批下发，进行异步渲染：

<table> 
 <thead> 
  <tr> 
   <th>推演方向</th> 
   <th>可行性分析</th> 
   <th>最终结论</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>敦促下游依赖提升服务性能</td> 
   <td>– 可行性低<br> – 投入产出比低</td> 
   <td><strong>无效推演</strong></td> 
  </tr> 
  <tr> 
   <td>将页面所需数据进行分批下发，进行异步渲染</td> 
   <td>核心数据均在首屏加载完毕</td> 
   <td><strong>无效推演</strong></td> 
  </tr> 
  <tr> 
   <td>异步加载数据</td> 
   <td>业务流程中存在较多无直接依赖数据可尝试并行加载</td> 
   <td><strong>有效推演</strong></td> 
  </tr> 
 </tbody> 
</table>

#### 并行加载技术方案的选取 ####

<table> 
 <thead> 
  <tr> 
   <th>技术方向</th> 
   <th>优点</th> 
   <th>缺点</th> 
   <th>最终选择</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Future</td> 
   <td>– JDK自带<br> – 简单易上手</td> 
   <td>只能通过阻塞或者轮询方式获取返回值</td> 
   <td>淘汰</td> 
  </tr> 
  <tr> 
   <td>CompletableFuture</td> 
   <td>– JDK自带<br> – 简单易上手<br> – 异步任务结束或者出错时，会自动回调某个方法<br> – 主线程设置回调后，无需关心异步任务的执行</td> 
   <td></td> 
   <td>入选</td> 
  </tr> 
  <tr> 
   <td>RxJava</td> 
   <td></td> 
   <td>– 学习成本高<br> – 公司内部暂未推广使用</td> 
   <td>淘汰</td> 
  </tr> 
  <tr> 
   <td>Reactor</td> 
   <td></td> 
   <td>– 学习成本高<br> – 公司内部暂未推广使用</td> 
   <td>淘汰</td> 
  </tr> 
 </tbody> 
</table>

### CompletableFuture 异步加载的实现 ###

CompletableFuture通过异步回调的方式，解决了开发过程中异步调用获取结果的难点。开发人员只需接触到CompletableFuture对象，以及CompletableFuture任务的执行结果，无需设计具体异步回调的实现，并可通过自定义线程池进一步优化任务的异步调用。

**CompletableFuture 的实例化**

CompletableFuture提供了四个静态方法来创建一个实例

![ebc4930f9b884a65a865c87d92f177c3.png][]

构造方法说明

<table> 
 <thead> 
  <tr> 
   <th>方法声明</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>public static CompletableFuture&lt;Void&gt; runAsync(Runnable runnable)</td> 
   <td>– 无需返回结果的异步调用<br> – 使用默认的线程池&nbsp;ForkJoinPool.commonPool()&nbsp;</td> 
  </tr> 
  <tr> 
   <td>public static CompletableFuture&lt;Void&gt; runAsync(Runnable runnable, Executor executor)</td> 
   <td>– 无需返回结果的异步调用<br> – 使用用户自定义的线程池</td> 
  </tr> 
  <tr> 
   <td>public static &lt;U&gt; CompletableFuture&lt;U&gt; supplyAsync(Supplier&lt;U&gt; supplier)</td> 
   <td>– 需要返回值的异步调用<br> – 使用默认的线程池&nbsp;ForkJoinPool.commonPool()&nbsp;</td> 
  </tr> 
  <tr> 
   <td>public static &lt;U&gt; CompletableFuture&lt;U&gt; supplyAsync(Supplier&lt;U&gt; supplier, Executor executor)</td> 
   <td>– 需要返回值的异步调用<br> – 使用用户自定义的线程池</td> 
  </tr> 
 </tbody> 
</table>

**代码示例**

*  无返回值且未指定线程池的

public static void main(String[] args) {
            CompletableFuture<Void> result = CompletableFuture.runAsync(
                    () -> testNoResult("JF") );
            int age = 11;
            System.out.println("JF今年的年纪是" + age);
        }
     
        public static void  testNoResult(String name) {
            System.out.println(name +" 调用了 没有返回值的异步方法");
        }

*  无返回值且指定了线程池

public static void main(String[] args) {
         TaskThreadPoolExecutor threads = new TaskThreadPoolExecutor(20, 100, 30,
                 TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(500));
     
         CompletableFuture<Void> result = CompletableFuture.runAsync(
                         () -> testNoResult("JF"), threads);
          
     }
     
     public static void  testNoResult(String name) {
         System.out.println(name +" 调用了 没有返回值,但是制定了线程池的的异步方法");
     }

*  有返回值且未指定线程池的

public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
     
         CompletableFuture<String> result = CompletableFuture.supplyAsync(
                         () -> testNoResult("JF"));
     
         String resultDesc = result.get(1000, TimeUnit.MILLISECONDS);
         System.out.println(resultDesc);
     }
     
      public static String  testNoResult(String name) {
         return (name +" 调用了 没有返回值,但是制定了线程池的的异步方法");
     }

*  有返回值且指定线程池

#### 获取任务执行执行结果 ####

<table> 
 <thead> 
  <tr> 
   <th>方法声明</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>public T get();</td> 
   <td>– 实现了Future接口的功能<br> – 会一直阻塞直到获取线程结果</td> 
  </tr> 
  <tr> 
   <td>public T get(long timeout, TimeUnit unit);</td> 
   <td>– 实现了Future接口的功能<br> – 指定时间内仍未获取到结果会抛出异常</td> 
  </tr> 
  <tr> 
   <td>public T getNow(T valueIfAbsent);</td> 
   <td>– 不会产生阻塞<br> – 如果任务未执行完成返回默认缺省值</td> 
  </tr> 
  <tr> 
   <td>public T join();</td> 
   <td>一直阻塞直到获取线程结果<br> – 唯一不同就是 join 方法不会抛出检查时异常&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

**触发任务完成**

<table> 
 <thead> 
  <tr> 
   <th>方法声明</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>public boolean complete(T value);</td> 
   <td>– 主动触发当前异步任务的完成，如果已完成，该方法返回false<br> – 如果未完成则返回true，且其他线程将传入的 T value 当做返回值</td> 
  </tr> 
  <tr> 
   <td>public boolean completeExceptionally(Throwable ex);</td> 
   <td>触发任务执行的异常</td> 
  </tr> 
 </tbody> 
</table>

**complete**

**completeExceptionally**

public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
            TaskThreadPoolExecutor threads = new TaskThreadPoolExecutor(20, 100, 30,
                    TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(500));
     
            CompletableFuture<String> result = CompletableFuture.supplyAsync(
                            () -> testNoResult("JF"), threads);
     
            BizException bizException = new BizException(BizErrorCodeEnum.PARAM_ERROR);
            boolean complete = result.completeExceptionally(bizException);
            System.out.println("并行是否完成？" + !complete +"   结果是  " + result.get());
        }
     
        public static String  testNoResult(String name)  {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                return "nothing";
     
            }
            return (name +" 调用了");
        }
     
    ===================================================================================
    输出结果
     
    Exception in thread "main" java.util.concurrent.ExecutionException: com.tuhu.boot.common.exceptions.BizException: 参数错误
        at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
        at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
        at com.tuhu.cl.install.agg.service.common.TaskConfig.main(TaskConfig.java:49)
    Caused by: com.tuhu.boot.common.exceptions.BizException: 参数错误
        at com.tuhu.cl.install.agg.service.common.TaskConfig.main(TaskConfig.java:47)

### **对任务执行结果进行下一步处理** ###

**只能接受任务正常执行后的回调**

<table> 
 <tbody> 
  <tr> 
   <th>方法声明</th> 
   <th>说明</th> 
   <th>代码示例</th> 
   <th>执行结果</th> 
  </tr> 
  <tr> 
   <td>public &lt;U&gt; CompletionStage&lt;U&gt; thenApply(Function&lt;? super T,? extends U&gt; fn);</td> 
   <td>拿到上一步任务执行的结果进行处理并且返回处理的结果</td> 
   <td> <pre><code>    CompletableFuture&lt;String&gt;&nbsp;completableFuture&nbsp;=&nbsp;
    CompletableFuture.supplyAsync(()&nbsp;-&gt;&nbsp;10)
    .thenApply(v&nbsp;-&gt;&nbsp;("上一步的执行的结果为："&nbsp;+&nbsp;v));
    System.out.println(completableFuture.join());</code></pre> <p></p> </td> 
   <td>上一步的执行的结果为：10</td> 
  </tr> 
  <tr> 
   <td>public CompletableFuture&lt;Void&gt; thenRun(Runnable action);</td> 
   <td>拿不到上一步任务执行的结果，但会执行Runnable接口的实现</td> 
   <td> <pre><code>   CompletableFuture&lt;Void&gt;&nbsp;completableFuture&nbsp;=&nbsp;
       CompletableFuture.supplyAsync(()&nbsp;-&gt;&nbsp;10)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;                  .thenRun(()&nbsp;
                                -&gt;&nbsp;System.out.println("上一步执行完成"));</code></pre> <p></p> </td> 
   <td>上一步执行完成</td> 
  </tr> 
  <tr> 
   <td>public CompletionStage&lt;Void&gt; thenAccept(Consumer&lt;? super T&gt; action);</td> 
   <td>可以拿到上一步任务执行的结果进行处理，但不需要返回处理的结果</td> 
   <td> <pre><code>CompletableFuture&lt;Void&gt;&nbsp;completableFuture&nbsp;=
CompletableFuture.supplyAsync(()&nbsp;-&gt;&nbsp;10).thenAccept(v&nbsp;-&gt;&nbsp;
System.out.println("上一步执行完成，结果为："&nbsp;+&nbsp;v));</code></pre> <p></p> </td> 
   <td>上一步执行完成，结果为：10</td> 
  </tr> 
 </tbody> 
</table>

**只能接收任务处理异常后的回调**

<table> 
 <tbody> 
  <tr> 
   <th style="width:173px;">方法声明</th> 
   <th style="width:103px;">说明</th> 
   <th style="width:337px;">代码示例</th> 
   <th style="width:54px;">执行结果</th> 
  </tr> 
  <tr> 
   <td style="width:173px;"> <p>public&nbsp;CompletionStage&lt;T&gt;&nbsp;</p> <p>exceptionally(</p> <p>Function&lt;Throwable, ? extends T&gt; fn);</p> </td> 
   <td style="width:103px;">– 任务执行过程中出现异常的时候，会回调exceptionally方法<br> – exceptionally能够将异常给吞了，并且fn的返回值会返回回去（有点降级的味道了）</td> 
   <td style="width:337px;"> <pre><code>CompletableFuture&lt;Integer&gt;&nbsp;completableFuture&nbsp;=&nbsp;
CompletableFuture.supplyAsync(()&nbsp;-&gt;&nbsp;{return&nbsp;100;}).
exceptionally(e&nbsp;-&gt;&nbsp;{System.out.println("出现异常了，
返回默认值");return&nbsp;110;});
System.out.println(completableFuture.join());</code></pre> <p></p> </td> 
   <td style="width:54px;">出现异常了，返回默认值<br> 100</td> 
  </tr> 
 </tbody> 
</table>

**能同时接收任务执行正常和异常的回调**

<table> 
 <tbody> 
  <tr> 
   <th style="width:205px;">方法声明</th> 
   <th style="width:108px;">说明</th> 
   <th style="width:281px;">代码示例</th> 
   <th style="width:88px;">执行结果</th> 
  </tr> 
  <tr> 
   <td style="width:205px;"> <p>public&nbsp;&lt;U&gt; CompletionStage&lt;U&gt;&nbsp;</p> <p>handle(BiFunction&lt;?&nbsp;super&nbsp;T,</p> <p>Throwable, ? extends U&gt; fn);&nbsp;</p> </td> 
   <td style="width:108px;">– 跟exceptionally有点像，但是exceptionally是出现异常才会回调，handle正常情况下也能回调<br> – 两者都有默认返回值</td> 
   <td style="width:281px;"></td> 
   <td style="width:88px;"></td> 
  </tr> 
  <tr> 
   <td style="width:205px;"> <p>public&nbsp;CompletionStage&lt;T&gt;&nbsp;</p> <p>whenComplete(BiConsumer&lt;?&nbsp;</p> <p>super&nbsp;T, ?&nbsp;super&nbsp;Throwable&gt; actin);</p> </td> 
   <td style="width:108px;">– 能接受正常或者异常的回调，并且不影响上个阶段的返回值，也就是主线程能获取到上个阶段的返回值<br> – 当出现异常时，无降级操作，主线程在获取执行异常任务的结果时，会抛出异常。</td> 
   <td style="width:281px;"> 
    <div> 
     <pre><code>CompletableFuture&lt;Integer&gt; completableFuture 
= CompletableFuture.
     supplyAsync(() -&gt; {
          int i = 1 / 0;
          return 10;
   }).whenComplete((r, e) -&gt; {
     System.out.println("whenComplete被调用了");
   });
    System.out.println(completableFuture.join());</code></pre> 
    </div> </td> 
   <td style="width:88px;"> <p></p> <p>whenComplete<br> 被调用了<br> Exception in thread&nbsp;<br> “main” java.util.<br> concurrent.<br> Completion</p> <p>Exception:</p> </td> 
  </tr> 
 </tbody> 
</table>

#### 结果合并 ####

<table> 
 <thead> 
  <tr> 
   <th>方法声明</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>public &lt;U,V&gt; CompletionStage&lt;V&gt; thenCombine(CompletionStage&lt;? extends U&gt; other, BiFunction&lt;? super T, ? super U, ? extends V&gt; fn);</td> 
   <td>组合两个 future，获取两个 future 的返回结果，并返回当前任务的返回值</td> 
  </tr> 
  <tr> 
   <td>public &lt;U&gt; CompletableFuture&lt;Void&gt; thenAcceptBoth(CompletionStage&lt;? extends U&gt; other, BiConsumer&lt;? super T, ? super U&gt; action)</td> 
   <td>组合两个 future，获取两个 future 任务的返回结果，然后处理任务，没有返回值</td> 
  </tr> 
  <tr> 
   <td>public CompletableFuture&lt;Void&gt; runAfterBoth(CompletionStage&lt;?&gt; other, Runnable action)</td> 
   <td>组合两个 future，不需要获取 future 的结果，只需两个 future 处理完任务后，处理该任务</td> 
  </tr> 
 </tbody> 
</table>

### 应用实践 ###

![a9667a33857e43a6bb906f102751626e.png][]

#### 代码示例 ####

@Resource("appHomeExecutor")
        private Executor executor;
    
        @Override
        public void initApp() {
            // 1:异步查询附近门店
            CompletableFuture<ShopInfoVo> shopCompletableFuture = CompletableFuture.supplyAsync(() -> {
                return new ShopInfoVo();
            }, executor);
            // 2:异步查询推荐商品
            CompletableFuture<ProductVo> productCompletableFuture = CompletableFuture.supplyAsync(() -> {
                return new ProductVo();
            }, executor);
            // 3:异步查询活动信息
            CompletableFuture<ActivityVo> activityVoCompletableFuture = CompletableFuture.supplyAsync(() -> {
                return new ActivityVo();
            }, executor);
            // 4:异步查询banner图
            CompletableFuture<BannerVo> bannerCompletableFuture = CompletableFuture.supplyAsync(() -> {
                return new BannerVo();
            }, executor);
    
            // 等待所有异步运算完成
            CompletableFuture.allOf(shopCompletableFuture, productCompletableFuture, activityVoCompletableFuture,
                bannerCompletableFuture);
    
            try {
                // 获取异步运算结果
                ShopInfoVo shopInfoVo = shopCompletableFuture.get();
                ProductVo productVo = productCompletableFuture.get();
                ActivityVo activityVo = activityVoCompletableFuture.get();
                BannerVo bannerVo = bannerCompletableFuture.get();
    
                // 组装响应数据
    
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

#### 实际效果 ####

--------------------

[参考][Link 1]

[ebc4930f9b884a65a865c87d92f177c3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/7ddd35d746c44b82bd3960804e743a7a.png
[a9667a33857e43a6bb906f102751626e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/ffca9e1fb2634607b2ec122a506c5867.png
[Link 1]: https://tech.tuhu.cn/?p=7448