CompletableFuture异步批处理

　　前言
　　在java8的环境下，CompletableFuture是非常受大家喜欢的api，其强大的异步编排能力是做应用服务的绝佳助手。这个api里面包含了大几十个方法，咱们不挨着展开，本篇默认的都是对这个api有一些基础的小伙伴。CompletableFutureCompletionStage代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段一个阶段的计算执行可以是一个Function，Consumer或者Runnable。比如：stage.thenApply(x -> square(x)).thenAccept(x -> System.out.print(x)).thenRun(() -> System.out.println())一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发在Java8中，CompletableFuture提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合 CompletableFuture 的方法。它可能代表一个明确完成的Future，也有可能代表一个完成阶段（ CompletionStage ），它支持在计算完成以后触发一些函数或执行某些动作。它实现了Future和CompletionStage接口
　　使用场景
　　现在有如下场景。前端传递一个用户对象的集合，拿到之后需要后端挨个处理集合中的对象，且每个对象的操作都非常地耗时，例如需要三秒左右，每一个处理完之后会返回一个结果，我需要汇总这些结果返回给前端。示意图大概如下：
　　挨个处理完每个对象之后，已经花费了10s，这还没算上此次其它的操作。假设业务上对这个接口的要求是5s，那么显然这样做就不满足要求了,那怎么办呢，也比较容易想到，每个处理对象的操作异步进行，最终把结果汇总下就行了。
　　思路如下：
　　结合CompletableFuture实现业务
　　新建user类：import lombok.Data;  /** * @author : wuwensheng * @date : 10:47 2021/12/13 */ @Data public class User {    private String name;     private int age;     public User(String name, int age) {        this.name = name;        this.age = age;    } } 复制代码
　　springboot整合测试：import com.teligen.PhoneApplication; import lombok.extern.slf4j.Slf4j; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner;  import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ThreadPoolExecutor; import java.util.function.BiFunction; import java.util.function.Supplier;  /**  * 异步批处理的类，可以有相关的许多变种  *  * @author : wuwensheng  * @date : 10:36 2021/12/13  */ @RunWith(SpringRunner.class) @SpringBootTest(classes = PhoneApplication.class) @Slf4j public class CompletableFutureTest {      @Autowired     private ThreadPoolExecutor customThreadPoolExecutor;      } 复制代码
　　customThreadPoolExecutor这个对象是我的线程池。
　　每个对user的处理都是一个CompletableFuture，如下：public CompletableFuture disposeUser(User user) {     return CompletableFuture.supplyAsync(new Supplier() {          @Override         public Integer get() {             log.info(＂Thread name:{}＂, Thread.currentThread().getName());             try {                 TimeUnit.SECONDS.sleep(2);             } catch (InterruptedException e) {                 e.printStackTrace();             }             return user.getAge() + 10;         }     }, customThreadPoolExecutor).handleAsync(new BiFunction() {         @Override         public Integer apply(Integer param, Throwable throwable) {             int result = param;             if (throwable == null) {                 result = param * 2;             } else {                 log.info(＂throwable is:{}＂, throwable.getMessage());             }             return result;         }     }); } 复制代码
　　这里处理每一个user都让当前线程沉睡了两秒，用来模拟处理业务所花费的时间。
　　继续编排下：@Test public void userTest() {     List users = new ArrayList<>();     users.add(new User(＂小明＂, 3));     users.add(new User(＂小红＂, 2));     users.add(new User(＂小芳＂, 18));     CompletableFuture[] completableFutures = users.stream().map(user -> {          return disposeUser(user);      }).toArray(CompletableFuture[]::new);     // 等待所有任务执行完     CompletableFuture.allOf(completableFutures).join();      for (CompletableFuture completableFuture : completableFutures) {         try {             log.info(＂result:{}＂, completableFuture.get());         } catch (Exception e) {             e.printStackTrace();         }     } } 复制代码
　　跑一下看看结果:
　　ok了，57秒开始处理，59秒处理完毕。任务在join之后的确是并行的。这是一种什么感觉呢，所有异步线程出去办事了，有一辆车等着它们回来，最后一个人回来的时候，那便发车。
　　咱们再验证下。当处理小芳的时候沉睡5秒，看下结果：
　　这次返回耗费了5秒左右，处理得最慢的那个线程决定了最终的返回时长，这也符合咱们的预期。 大家在处理集合数据并且每一条的处理都比较耗时的话，可以考虑这个手法。
　　作者：119_115_104_104_201
　　链接：https://juejin.cn/post/7041087572461748232
　　来源：稀土掘金
　　著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。