CompletableFuture实现多个互不相干的方法并行调用

柔光的暖阳◎ 2023-09-29 15:17 8阅读 0赞

### 背景 ###

在系统的后台管理页面中，有一些数据的统计需求，往往是前端发一个请求，后端需要调用多个服务来进行数据的查询，然后组装给前端；如果数据量特别大，假设每个数据的查询需要1秒，有几个数据查询就需要多少秒，执行时间随着统计量上升而上升。

经分析，各个数据的查询是独立的，可以考虑改为并行统计，最后再合并，耗时只会是最耗时的那个服务，再加上并行计算额外的小部分耗时。

![57c099ea5146452e8fdd851ba8a5f327.png][]

### 串行处理 ###

之前的串行处理，就是在一个方法中，挨个调用其他方法，其他方法可能在同一个服务上，也可能在不同的微服务上，耗时等于各个方法调用之和。

public static Map dataStatistics() {
            Map<String, Object> data = new HashMap<>();
            data.put("userNum", countUserNum());
            data.put("companyNum", countCompanyNum());
            data.put("workNum", countWorkNum());
            data.put("taskNum", countTaskNum());
            data.put("courseNum", countCourseNum());
            data.put("missionNum", countMissonNum());
            data.put("loginNum", countLoginNum());
    
            return data;
        }

### 并行处理 ###

使用CompletableFuture的supplyAsync异步调用方法，相当于每个方法单独起一个线程去执行

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> countUserNum(), executorService);
            CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> countCompanyNum(), executorService);
            CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> countWorkNum(), executorService);
            CompletableFuture<Integer> future4 = CompletableFuture.supplyAsync(() -> countTaskNum(), executorService);
            CompletableFuture<Integer> future5 = CompletableFuture.supplyAsync(() -> countCourseNum(), executorService);
            CompletableFuture<Integer> future6 = CompletableFuture.supplyAsync(() -> countMissonNum(), executorService);
            CompletableFuture<Integer> future7 = CompletableFuture.supplyAsync(() -> countLoginNum(), executorService);

最后用 CompletableFuture.allOf方法等待所有方法执行完毕，然后取出结果，这样所有接口的耗时就取决于最慢的那个服务/方法。

CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2, future3, future4, future5, future6, future7);

###  ###

### 测试代码 ###

完整的测试代码，构建一个数据查询场景，每个数据查询需要1秒

package com.zubus.admin;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Random;
    import java.util.concurrent.*;
    
    public class LeanCompletableFuture {
        static ExecutorService executorService = Executors.newFixedThreadPool(7);
    
    
        public static Map dataStatisticsParallel() {
    
            CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> countUserNum(), executorService);
            CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> countCompanyNum(), executorService);
            CompletableFuture<Integer> future3 = CompletableFuture.supplyAsync(() -> countWorkNum(), executorService);
            CompletableFuture<Integer> future4 = CompletableFuture.supplyAsync(() -> countTaskNum(), executorService);
            CompletableFuture<Integer> future5 = CompletableFuture.supplyAsync(() -> countCourseNum(), executorService);
            CompletableFuture<Integer> future6 = CompletableFuture.supplyAsync(() -> countMissonNum(), executorService);
            CompletableFuture<Integer> future7 = CompletableFuture.supplyAsync(() -> countLoginNum(), executorService);
    
    
            CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2, future3, future4, future5, future6, future7);
            try {
                allOf.get();
    
                Map<String, Object> data = new HashMap<>();
                data.put("userNum", future1.get());
                data.put("companyNum", future2.get());
                data.put("workNum", future3.get());
                data.put("taskNum", future4.get());
                data.put("courseNum", future5.get());
                data.put("missionNum", future6.get());
                data.put("loginNum", future7.get());
    
                return data;
    
    
            } catch (InterruptedException exception) {
    
            } catch (ExecutionException exception) {
    
            } finally {
                executorService.shutdown();
            }
    
            return null;
        }
    
    
        public static void main(String[] args) {
    
    
            System.out.println("---串行开始----");
    
            long start = System.currentTimeMillis();
            Map map = dataStatistics();
    
            map.forEach((key, value) -> System.out.println(key + ":" + value));
    
            long end = System.currentTimeMillis();
    
    
            map.forEach((key, value) -> System.out.println(key + ":" + value));
            System.out.println("---串行结束-- 耗时--" + (end - start));
    
    
            System.out.println();
            System.out.println();
            System.out.println();
            System.out.println();
    
            System.out.println("---并行开始----");
            long start1 = System.currentTimeMillis();
            Map map1 = dataStatisticsParallel();
    
            long end1 = System.currentTimeMillis();
            map1.forEach((key, value) -> System.out.println(key + ":" + value));
            System.out.println("---并行结束-- 耗时--" + (end1 - start1));
    
    
        }
    
        /**
         * 首页大屏数据统计，每个模拟耗时1秒
         *
         * @return
         */
        public static Map dataStatistics() {
            Map<String, Object> data = new HashMap<>();
            data.put("userNum", countUserNum());
            data.put("companyNum", countCompanyNum());
            data.put("workNum", countWorkNum());
            data.put("taskNum", countTaskNum());
            data.put("courseNum", countCourseNum());
            data.put("missionNum", countMissonNum());
            data.put("loginNum", countLoginNum());
    
            return data;
        }
    
        private static Integer countLoginNum() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
            Integer countLoginNum = new Random().nextInt(1000);
            return countLoginNum == null ? 0 : countLoginNum * 5;
        }
    
        private static Integer countMissonNum() {
    
            try {
                Thread.sleep(1000);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
            Integer countMissonNum = new Random().nextInt(1000);
            return countMissonNum == null ? 0 : countMissonNum;
        }
    
        private static Integer countCourseNum() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
            Integer countCourseNum = new Random().nextInt(1000);
            return countCourseNum == null ? 0 : countCourseNum;
        }
    
        private static Integer countTaskNum() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
            Integer countTaskNum = new Random().nextInt(1000);
            return countTaskNum == null ? 0 : countTaskNum;
        }
    
        private static Integer countWorkNum() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
            Integer countWorkNum = new Random().nextInt(1000);
            return countWorkNum == null ? 0 : countWorkNum;
        }
    
        private static Integer countCompanyNum() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
    
            Integer countCompanyNum = new Random().nextInt(1000);
            return countCompanyNum == null ? 0 : countCompanyNum;
        }
    
        private static Integer countUserNum() {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
    
            Integer countUserNum = new Random().nextInt(1000);
            return countUserNum == null ? 0 : countUserNum;
        }
    
    }

### 测试结果 ###

效果还是非常明显的，串行查询如预测的那样，就是每个方法执行时间的总和，而并行的几乎和一个方法执行的时间相同。

![5877e2db88184abbaf6c31c657b936ff.png][]

### 注意 ###

如果数据量特别小，或者串行系统响应速度也非常的快，就没必要改为并行，因为并行线程调度本身也是有开销的；好比挖掘机虽然比铲子快，可地上就三锹土，挖掘机一下给你搞定，你用铲子三下也能搞定，但你把挖掘机开过来却要半小时。

[57c099ea5146452e8fdd851ba8a5f327.png]: https://img-blog.csdnimg.cn/57c099ea5146452e8fdd851ba8a5f327.png
[5877e2db88184abbaf6c31c657b936ff.png]: https://img-blog.csdnimg.cn/5877e2db88184abbaf6c31c657b936ff.png