【Java多线程】CompletableFuture 异步多线程

谁践踏了优雅 2024-03-23 21:52 102阅读 0赞

### 1. 回顾 Future ###

一些业务场景我们需要使用`多线程异步`执行任务，加快任务执行速度。

`JDK5`新增了`Future`接口，用于描述一个异步计算的结果。

虽然 Future 以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，我们必须使用`Future.get()`的方式`阻塞`调用线程，或者使用`轮询`方式判断 `Future.isDone` 任务是否结束，再获取结果。

这两种处理方式都不是很优雅，相关代码如下:

@Test 
 public void testFuture() throws ExecutionException, InterruptedException {
 
 ExecutorService executorService = Executors.newFixedThreadPool(5); 
 Future<String> future = executorService.submit(() -> {
 
 Thread.sleep(2000); 
 return "hello"; 
 }); 
 System.out.println(future.get()); 
 System.out.println("end"); 
 }

与此同时，`Future`无法解决`多个异步任务需要相互依赖`的场景，简单点说就是，主线程需要等待子线程任务执行完毕之后在进行执行，这个时候你可能想到了`CountDownLatch`，没错确实可以解决，代码如下。

这里定义两个`Future`，第一个通过用户id获取用户信息，第二个通过商品id获取商品信息。

@Test 
 public void testCountDownLatch() throws InterruptedException, ExecutionException {
 
 ExecutorService executorService = Executors.newFixedThreadPool(5); 
 CountDownLatch downLatch = new CountDownLatch(2); 
 long startTime = System.currentTimeMillis(); 
 Future<String> userFuture = executorService.submit(() -> {
 
 //模拟查询商品耗时500毫秒 
 Thread.sleep(500); 
 downLatch.countDown(); 
 return "用户A"; 
 }); 
 
 Future<String> goodsFuture = executorService.submit(() -> {
 
 //模拟查询商品耗时500毫秒 
 Thread.sleep(400); 
 downLatch.countDown(); 
 return "商品A"; 
 }); 
 
 downLatch.await(); 
 //模拟主程序耗时时间 
 Thread.sleep(600); 
 System.out.println("获取用户信息:" + userFuture.get()); 
 System.out.println("获取商品信息:" + goodsFuture.get()); 
 System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms"); 
 
 }

「运行结果」

> 获取用户信息:用户A  
> 获取商品信息:商品A  
> 总共用时1110ms  
> 从运行结果可以看出结果都已经获取，而且如果我们不用异步操作，执行时间应该是:500+400+600 = 1500，用异步操作后实际只用1110。

但是`Java8`以后这就不在认为是一种优雅的解决方式，接下来了解下`CompletableFuture`的使用。

### 2. CompletableFuture ###

`JDK8`之后，提供了`CompletableFuture`实现异步线程。不推荐再使用`Future`

**通过`CompletableFuture`实现上面示例**

@Test 
 public void testCompletableInfo() throws InterruptedException, ExecutionException {
 
 long startTime = System.currentTimeMillis(); 
 
 //调用用户服务获取用户基本信息 
 CompletableFuture<String> userFuture = CompletableFuture.supplyAsync(() -> 
 //模拟查询商品耗时500毫秒 
 {
 
 try {
 
 Thread.sleep(500); 
 } catch (InterruptedException e) {
 
 e.printStackTrace(); 
 } 
 return "用户A"; 
 }); 
 
 //调用商品服务获取商品基本信息 
 CompletableFuture<String> goodsFuture = CompletableFuture.supplyAsync(() -> 
 //模拟查询商品耗时500毫秒 
 {
 
 try {
 
 Thread.sleep(400); 
 } catch (InterruptedException e) {
 
 e.printStackTrace(); 
 } 
 return "商品A"; 
 }); 
 
 System.out.println("获取用户信息:" + userFuture.get()); 
 System.out.println("获取商品信息:" + goodsFuture.get()); 
 
 //模拟主程序耗时时间 
 Thread.sleep(600); 
 System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms"); 
 }

**\[运行结果\]**

> 获取用户信息:用户A  
> 获取商品信息:商品A  
> 总共用时1112ms

通过`CompletableFuture`可以很轻松的实现`CountDownLatch`的功能  
当然，CompletableFuture还有其他许多新功能：比如可以实现：任务1执行完了再执行任务2,甚至任务1执行的结果，作为任务2的入参数等等强大功能，下面就来学学CompletableFuture的API。

**CompletableFuture创建方式**

#### 1、常用的4种创建方式 ####

CompletableFuture源码中有四个静态方法用来执行异步任务

public static CompletableFuture supplyAsync(Supplier supplier){
 ..} 
 public static CompletableFuture supplyAsync(Supplier supplier,Executor executor){
 ..} 
 public static CompletableFuture<Void> runAsync(Runnable runnable){
 ..} 
 public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor){
 ..}

一般我们用上面的静态方法来创建`CompletableFuture`，  
区别:  
`supplyAsync`执行任务，支持返回值。  
`runAsync`执行任务，没有返回值。  
`supplyAsync`方法

//使用默认内置线程池ForkJoinPool.commonPool()，根据supplier构建执行任务 
 public static CompletableFuture supplyAsync(Supplier supplier) 
 //自定义线程，根据supplier构建执行任务 
 public static CompletableFuture supplyAsync(Supplier supplier, Executor executor) 
 「runAsync方法」
 
 //使用默认内置线程池ForkJoinPool.commonPool()，根据runnable构建执行任务 
 public static CompletableFuture<Void> runAsync(Runnable runnable) 
 //自定义线程，根据runnable构建执行任务 
 public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)

#### 2、结果获取的4种方式 ####

对于结果的获取CompltableFuture类提供了四种方式

//方式一  
    public T get()  
    //方式二  
    public T get(long timeout, TimeUnit unit)  
    //方式三  
    public T getNow(T valueIfAbsent)  
    //方式四  
    public T join()

说明：

`get()`和`get(long timeout, TimeUnit unit)` => 在Future中就已经提供了，后者提供超时处理，如果在指定时间内未获取结果将抛出超时异常  
`getNow()` => 立即获取结果不阻塞，结果计算已完成将返回结果或计算过程中的异常，如果未计算完成将返回设定的valueIfAbsent值  
`join()` => 方法里不会抛出异常  
示例：

@Test 
 public void testCompletableGet() throws InterruptedException, ExecutionException {
 
 
 CompletableFuture<String> cp1 = CompletableFuture.supplyAsync(() -> {
 
 try {
 
 Thread.sleep(1000); 
 } catch (InterruptedException e) {
 
 e.printStackTrace(); 
 } 
 return "商品A"; 
 }); 
 
 // getNow方法测试 
 System.out.println(cp1.getNow("商品B")); 
 
 //join方法测试 
 CompletableFuture<Integer> cp2 = CompletableFuture.supplyAsync((() -> 1 / 0)); 
 System.out.println(cp2.join()); 
 System.out.println("-----------------------------------------------------"); 
 //get方法测试 
 CompletableFuture<Integer> cp3 = CompletableFuture.supplyAsync((() -> 1 / 0)); 
 System.out.println(cp3.get()); 
 }

**「运行结果」**

> 第一个执行结果为 「商品B」，因为要先睡上1秒结果不能立即获取  
> join方法获取结果方法里不会抛异常，但是执行结果会抛异常，抛出的异常为CompletionException  
> get方法获取结果方法里将抛出异常，执行结果抛出的异常为ExecutionException

### 3. CountDownLatch ###

`CountDownLatch`可以使一个获多个线程等待其他线程各自执行完毕后再执行。

> CountDownLatch 定义了一个计数器，和一个阻塞队列， 当计数器的值递减为0之前，阻塞队列里面的线程处于挂起状态，当计数器递减到0时会唤醒阻塞队列所有线程，这里的计数器是一个标志，可以表示一个任务一个线程，也可以表示一个倒计时器，CountDownLatch可以解决那些一个或者多个线程在执行之前必须依赖于某些必要的前提业务先执行的场景。

**CountDownLatch 常用方法说明**

CountDownLatch(int count); //构造方法，创建一个值为count 的计数器。
    
    await();//阻塞当前线程，将当前线程加入阻塞队列。
    
    await(long timeout, TimeUnit unit);//在timeout的时间之内阻塞当前线程,时间一过则当前线程可以执行，
    
    countDown();//对计数器进行递减1操作，当计数器递减至0时，当前线程会去唤醒阻塞队列里的所有线程。

**举例：用CountDownLatch 来优化我们的报表统计**

> 运营系统有统计报表、业务为统计每日的用户新增数量、订单数量、商品的总销量、总销售额…等多项指标统一展示出来，因为数据量比较大，统计指标涉及到的业务范围也比较多，所以这个统计报表的页面一直加载很慢，所以需要对统计报表这块性能需进行优化。

**问题分析:**  
统计报表页面涉及到的统计指标数据比较多，每个指标需要单独的去查询统计数据库数据，单个指标只要几秒钟，但是页面的指标有10多个，所以整体下来页面渲染需要将近一分钟。

**解决方案:**  
任务时间长是因为统计指标多，而且指标是串行的方式去进行统计的，我们只需要考虑把这些指标从串行化的执行方式改成并行的执行方式，那么整个页面的时间的渲染时间就会大大的缩短， 如何让多个线程同步的执行任务，我们这里考虑使用多线程，每个查询任务单独创建一个线程去执行，这样每个统计指标就可以并行的处理了。

**要求:**  
因为主线程需要每个线程的统计结果进行聚合，然后返回给前端渲染，所以这里需要提供一种机制让主线程等所有的子线程都执行完之后再对每个线程统计的指标进行聚合。 这里我们使用CountDownLatch 来完成此功能。

**模拟代码**

> 1、分别统计4个指标用户新增数量、订单数量、商品的总销量、总销售额；  
> 2、假设每个指标执行时间为3秒。如果是串行化的统计方式那么总执行时间会为12秒。  
> 3、我们这里使用多线程并行，开启4个子线程分别进行统计  
> 4、主线程等待4个子线程都执行完毕之后，返回结果给前端。

//用于聚合所有的统计指标
        private static Map map=new HashMap();
        //创建计数器，这里需要统计4个指标
        private static CountDownLatch countDownLatch=new CountDownLatch(4);
    
        public static void main(String[] args) {
        
            //记录开始时间
            long startTime=System.currentTimeMillis();
    
            Thread countUserThread=new Thread(new Runnable() {
        
                public void run() {
        
                    try {
        
                        System.out.println("正在统计新增用户数量");
                        Thread.sleep(3000);//任务执行需要3秒
                        map.put("userNumber",1);//保存结果值
                        //countDownLatch.countDown();//标记已经完成一个任务
                        System.out.println("统计新增用户数量完毕");
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    } finally{
        
                    	countDownLatch.countDown();//标记已经完成一个任务
                    }
    
                }
            });
            Thread countOrderThread=new Thread(new Runnable() {
        
                public void run() {
        
                    try {
        
                        System.out.println("正在统计订单数量");
                        Thread.sleep(3000);//任务执行需要3秒
                        map.put("countOrder",2);//保存结果值
                        System.out.println("统计订单数量完毕");
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    } finally {
        
                    	countDownLatch.countDown();//标记已经完成一个任务
                    }
                }
            });
    
            Thread countGoodsThread=new Thread(new Runnable() {
        
                public void run() {
        
                    try {
        
                        System.out.println("正在商品销量");
                        Thread.sleep(3000);//任务执行需要3秒
                        map.put("countGoods",3);//保存结果值
                        System.out.println("统计商品销量完毕");
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    } finally {
        
                    	countDownLatch.countDown();//标记已经完成一个任务
                   	}
                }
            });
    
            Thread countmoneyThread=new Thread(new Runnable() {
        
                public void run() {
        
                    try {
        
                        System.out.println("正在总销售额");
                        Thread.sleep(3000);//任务执行需要3秒
                        map.put("countmoney",4);//保存结果值
                        System.out.println("统计销售额完毕");
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    } finally {
        
                    	countDownLatch.countDown();//标记已经完成一个任务
                   	}
                }
            });
            
            //启动子线程执行任务
            countUserThread.start();
            countGoodsThread.start();
            countOrderThread.start();
            countmoneyThread.start();
    
            try {
        
                //主线程等待所有统计指标执行完毕
                countDownLatch.await();
                long endTime=System.currentTimeMillis();//记录结束时间
                System.out.println("------统计指标全部完成--------");
                System.out.println("统计结果为："+map.toString());
                System.out.println("任务总执行时间为"+(endTime-startTime)/1000+"秒");
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }

注意：`countDownLatch.countDown();`放在`finally`中，防止线程异常把机器卡死

**「运行结果」**  
![在这里插入图片描述][8d3c54067d7944f0a406ed8b6226a432.png]

> 参考链接  
> [https://zhuanlan.zhihu.com/p/647743286][https_zhuanlan.zhihu.com_p_647743286]  
> [https://zhuanlan.zhihu.com/p/95835099?utm\_id=0][https_zhuanlan.zhihu.com_p_95835099_utm_id_0]

[8d3c54067d7944f0a406ed8b6226a432.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/77f9741aec184b73900410c34c99e948.png
[https_zhuanlan.zhihu.com_p_647743286]: https://zhuanlan.zhihu.com/p/647743286
[https_zhuanlan.zhihu.com_p_95835099_utm_id_0]: https://zhuanlan.zhihu.com/p/95835099?utm_id=0