CompletableFuture实战
一、介绍
CompletableFuture 是 Java 8 中引入的一个接口实现类，它位于 java.util.concurrent 包中。它是一个可以表示异步计算的结果的类，提供了比 Future 更加强大的功能来组合和执行异步操作。CompletableFuture 提供了构建复杂异步计算的能力，而不需要显式地管理线程。
二、依赖引入
CompletableFuture内置于JUC包中，不需要额外引入其他依赖。本文还用到Hutools用于计时器、Lombok用于简化get、set方法和自动生成构造函数
<!--hutool依赖--><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.8.22</version></dependency><!--junit测试依赖--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency><!--lombok--><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.20</version></dependency>
三、前置准备
3.1、实体类
Order
import lombok.AllArgsConstructor;
import lombok.Data;@Data
@AllArgsConstructor
public class Order {private Long id;private Long productId;private Long userId;
}Product
import lombok.AllArgsConstructor;
import lombok.Data;@Data
@AllArgsConstructor
public class Product {private Long id;private String name;private Double price;
}User
import lombok.AllArgsConstructor;
import lombok.Data;@Data
@AllArgsConstructor
public class User {private Long id;private String name;
}3.2、service层
UserService
import cn.hutool.core.date.DateUtil;
import com.demo.future.entity.User;import java.util.ArrayList;
import java.util.List;public class UserService {private List<User> list = new ArrayList<>();public UserService() {list.add(new User(11L,"小明"));list.add(new User(22L,"小王"));list.add(new User(33L,"张三"));list.add(new User(44L,"李四"));}public List<User> searchUser() {System.out.println(DateUtil.now()+"执行UserService的searchUser方法");try {Thread.sleep(3000);} catch (InterruptedException e) {throw new RuntimeException(e);}return list;}
}OrderService
import cn.hutool.core.date.DateUtil;
import com.demo.future.entity.Order;import java.util.ArrayList;
import java.util.List;public class OrderService {private List<Order> list = new ArrayList<>();public OrderService() {list.add(new Order(111L,1L,11L));list.add(new Order(222L,2L,22L));list.add(new Order(333L,3L,33L));list.add(new Order(444L,4L,44L));}public List<Order> searchOrder() {System.out.println(DateUtil.now()+"执行orderService的searchOrder方法");try {Thread.sleep(4000);} catch (InterruptedException e) {throw new RuntimeException(e);}return list;}
}ProductService
import cn.hutool.core.date.DateUtil;
import com.demo.future.entity.Product;import java.util.ArrayList;
import java.util.List;public class ProductService {private List<Product> list = new ArrayList<>();public ProductService() {list.add(new Product(1L,"苹果",52.4));list.add(new Product(2L,"橘子",100.5));list.add(new Product(3L,"香蕉",245.1));list.add(new Product(4L,"草莓",65.1));}public List<Product> searchProduct() {System.out.println("执行ProductService的searchProduct方法");try {Thread.sleep(5000);} catch (InterruptedException e) {throw new RuntimeException(e);}return list;}
}四、说明
4.1、传统串行化运行
我们在工作中是不是会经常这样写代码，分别new每个service，然后执行。执行顺序就看哪个方法在前面，就先执行，如果方法陷入阻塞，后续的方法就要等待前面的方法执行完成。这样就是串行化执行，执行顺序是searchUser()->searchProduct()->searchOrder()。
TimeInterval timer = DateUtil.timer();
UserService userService = new UserService();
ProductService productService = new ProductService();
OrderService orderService = new OrderService();
//阻塞了3秒
userService.searchUser();
//阻塞了4秒
productService.searchProduct();
//阻塞了5秒
orderService.searchOrder();
System.out.println("本次执行一共花费" + timer.interval() / 1000 + "秒");
运行这段代码后，输出如下图
4.2、异步执行所有任务
我们这次用到CompletableFuture.supplyAsync()，这是一个静态方法，用于创建一个 CompletableFuture 实例，它异步执行一个 Supplier 函数，并返回其结果。这个方法通常用于启动一个异步任务，该任务最终会产生一个结果。
方法签名如下：
/** 简单翻译就是这个方法会返回一个异步完成的新CompletableFuture* 他会在共享的commonPool()线程池里面运行* Returns a new CompletableFuture that is asynchronously completed* by a task running in the {@link ForkJoinPool#commonPool()} with* the value obtained by calling the given Supplier.** @param supplier a function returning the value to be used* to complete the returned CompletableFuture* @param <U> the function's return type* @return the new CompletableFuture*/public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {return asyncSupplyStage(asyncPool, supplier);} /** 这个也是一样的，只是多了executor参数，用于执行 supplier 的 Executor* Returns a new CompletableFuture that is asynchronously completed* by a task running in the given executor with the value obtained* by calling the given Supplier.** @param supplier a function returning the value to be used* to complete the returned CompletableFuture* @param executor the executor to use for asynchronous execution* @param <U> the function's return type* @return the new CompletableFuture*/public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor) {return asyncSupplyStage(screenExecutor(executor), supplier);}
我们引入CompletableFuture，来尝试第一种方法，异步执行所有任务。即userService、productService和orderService同时调用方法。
TimeInterval timer = DateUtil.timer();UserService userService = new UserService();
ProductService productService = new ProductService();
OrderService orderService = new OrderService();CompletableFuture<List<User>> userFuture = CompletableFuture.supplyAsync(userService::searchUser);
CompletableFuture<List<Product>> productFuture = CompletableFuture.supplyAsync(productService::searchProduct);
CompletableFuture<List<Order>> orderFuture = CompletableFuture.supplyAsync(orderService::searchOrder);
CompletableFuture.allOf(userFuture, productFuture, orderFuture);System.out.println(userFuture.get());
System.out.println(productFuture.get());
System.out.println(orderFuture.get());System.out.println("本次执行一共花费" + timer.interval() / 1000 + "秒");
执行结果如下，我们可以看到是同时调用了三个方法，但是由于每个方法阻塞的时间不同，首先是用户信息先返回，然后是产品，最后是订单。因为最大阻塞的时间是5秒，所以执行的时间就是5秒。
4.3 任务编排
CompletableFuture还有更高级的使用，就是任务编排。顾名思义就是先等A任务执行完成，再去执行BC任务。
执行流程
方法介绍
/** 返回一个已完成的新CompletableFuture，相当于直接执行方法* Returns a new CompletableFuture that is already completed with* the given value.** @param value the value* @param <U> the type of the value* @return the completed CompletableFuture*/public static <U> CompletableFuture<U> completedFuture(U value) {return new CompletableFuture<U>((value == null) ? NIL : value);}
代码
TimeInterval timer = DateUtil.timer();
UserService userService = new UserService();
ProductService productService = new ProductService();
OrderService orderService = new OrderService();CompletableFuture<List<User>> userFuture = CompletableFuture.completedFuture(userService.searchUser());
//得到执行结果，执行完成后再继续做下面的操作
System.out.println(userFuture.get());//异步执行searchProduct()和searchOrder()
CompletableFuture<List<Product>> productFuture = CompletableFuture.supplyAsync(productService::searchProduct);
CompletableFuture<List<Order>> orderFuture = CompletableFuture.supplyAsync(orderService::searchOrder);
CompletableFuture.allOf(productFuture, orderFuture);System.out.println(productFuture.get());
System.out.println(orderFuture.get());System.out.println("本次执行一共花费" + timer.interval() / 1000 + "秒");
执行结果
最终执行时间就是A的3秒+C的5秒=8秒
4.4 任务编排2
我们在实际项目中，会遇到要A和B都是异步执行，但是要等A和B都执行完毕才去执行C，这时候我们就会用到这一种任务编排。
执行流程
方法介绍
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) {return uniApplyStage(null, fn);}
代码
TimeInterval timer = DateUtil.timer();UserService userService = new UserService();
ProductService productService = new ProductService();
OrderService orderService = new OrderService();CompletableFuture<List<User>> userFuture = CompletableFuture.supplyAsync(userService::searchUser);
CompletableFuture<List<Product>> productFuture = CompletableFuture.supplyAsync(productService::searchProduct);//当productFuture执行完成后，才会运行thenApply里面的函数
CompletableFuture<List<Order>> orderFuture = productFuture.thenApply((products) -> orderService.searchOrder());CompletableFuture.allOf(userFuture, productFuture);System.out.println(userFuture.get());
System.out.println(productFuture.get());
System.out.println(orderFuture.get());System.out.println("本次执行一共花费" + timer.interval() / 1000 + "秒");
执行结果
最终执行时间=searchProduct()的5秒+searchOrder()的四秒
