用线程池、缓存技术提升数十倍接口性能

比眉伴天荒 2022-09-03 11:25 163阅读 0赞

刚刚，现网发来急报，某新增数据接口请求超时了，功能无法使用。不可能吧！该系统已经上线多年，为何现在才发现此问题呢？再次请求改接口，分析日志发现，虽然前端请求已经超时，但是后端服务还在坚持不懈、任劳任怨的努力干活。一看处理时间，我靠，25分钟左右才处理完成，是可忍，孰不可忍！领导大手一挥，小C，这个问题你速速解决掉。事关现网，接到命令的我丝毫不敢怠慢，马上开始了本次攻关之旅

首先，询问了引发下此故障的原因，是因为操作人员在本次操作中导入了10000数据，不难想象，问题肯定就出在这10000条数据上。

接着，进行后端代码逻辑分析，伪代码如下。

public Response saveOrder(OrderRequest request) { 
        //请求校验
        validateRequest(request);
        //处理一部分业务逻辑
        preHandleService();
        //list.getSize()=10000
        List<Device> list= request.getData;
        for (int i = 0; i < list.size(); i++) { 
            ...
            dbQuery(condition);
            ...
            deviceService.dbInsert(device);;
            ...
            //整个过程代码有一百多行
        }
        //处理一部分逻辑
        postHandleService();
        //封装返回结果
        return wrapperResonse();
    }

由代码可以看到，后端在拿到这10000条数据后，对这些数据进行了for循环处理操作，for循环中还有db  
操作，打印了下每次for循环处理的时间，我靠，单个循环就花了150ms左右，那么10000条数据的总时间都快半小时了，这不超时才怪。然后，第一反应是缩小for循环时间，但是，想要优化到整个处理过程在一分钟之内吧，那么单个循环时间就要控制在6ms左右，这怎么可能，因为整个循环的代码行数就有一百多行，而且中间还有db操作呢。

怎么办呢？灵光突然一闪！对了，我可以用多线程技术对for循环中操作封装成task进行处理，使用new线程的方式处理任务肯定行不通，那就使用线程池吧，由于前端需要同步返回处理结果，那么还得使用线程池的submit方法，到时候对返回结果进行`get()`阻塞。

说干就干，找了下，系统中有定义好的线程池，长这样：

ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 80, 1, TimeUnit.MINUTES,
                new LinkedBlockingDeque<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());

10个核心线程，80个最大线程，活跃1分钟，队列最大放1000个任务，拒绝后再次使用本线程执行任务，也就是说，任务会保证执行。但是，感觉这个队列大小有点问题，我有一万条数据呢，可能会导致频繁的拒绝任务，所以我就修改了下参数。

ThreadPoolExecutor executor = new ThreadPoolExecutor(80, 80, 2, TimeUnit.MINUTES,
                new LinkedBlockingDeque<>(20000), new ThreadPoolExecutor.CallerRunsPolicy());

接着，将for循环拆出，作为任务提交给线程池，修改后代码：

public Response saveOrder(OrderRequest request) { 
        //请求校验
        validateRequest(request);
        //处理一部分业务逻辑
        preHandleService();
        //list.getSize()=10000
        List<Device> list= request.getData;
        //收集任务结果
        List<Future<Boolean>> saveResults = new ArrayList<>();
        for (int i = 0; i < list.size(); i++) { 
            Future<Boolean> save = executor.submit(() -> doSave());
            saveResults.add(save);
        }
        //阻塞结果
        saveResults.forEach(s-> { 
            try { 
                s.get();
            } catch (Exception e) { 
                log.error("Handle save data error.", e);
            }
        });
        //处理一部分逻辑
        postHandleService();
        //封装返回结果
        return wrapperResonse();
    }
    
    private boolean doSave() { 
        ...
        dbQuery(condition);
            ...
        deviceService.dbInsert(device);;
            ...
        //整个过程代码有一百多行
    }

还不错，代码顺利改造完成了，我急不可耐的进行了10000条数据测试，还不错，整个过程平均时间为67秒。现网环境肯定比我的破机器好，所以优化工作基本完成了。但是，本着精益求精的精神，我觉得还可以再继续优化。

随后打印了for循环里层insert时间，发现，整个过程的耗时大部分在这里了，那么为何不想想怎么将for循环插入改成batch insert呢，说干就干。

首先，我需要讲这些数据收集起来。但是插入操作是调用另一个微服务的进行的，这有点不好弄了。该怎么将这些数据收集起来，然后一次性批量插入呢？

灵光突然一闪！我可以在外层定义一个本次批量插入的唯一标识batchSn,就用uuid吧，然后将batch作为redis缓存的key，使用redis的set数据结构将，每次插入的数据device缓存起来，然后，循环结束后，调用一个执行批量操作的方法，从缓存中取出数据，批量插入db，最后记住要清除本次缓存值，释放空间！太完美了！伪代码如下：

public Response saveOrder(OrderRequest request) { 
        //请求校验
        validateRequest(request);
        //处理一部分业务逻辑
        preHandleService();
        //list.getSize()=10000
        List<Device> list = request.getData;
        List<Future<Boolean>> saveResults = new ArrayList<>();
        for (int i = 0; i < list.size(); i++) { 
            Future<Boolean> save = executor.submit(() -> doSave());
            saveResults.add(save);
        }
        //阻塞结果
        saveResults.forEach(s -> { 
            try { 
                s.get();
            } catch (Exception e) { 
                log.error("Handle save data error.", e);
            }
        });
        //进行批量操作指令
        doSaveBatch(UUID.randomUUID().toString().replace("-", ""));
        //处理一部分逻辑
        postHandleService();
        //封装返回结果
        return wrapperResonse();
    }
    
    private void doSaveBatch(String batchSn) { 
        try { 
            List<Device> list = jedisClient.members(batchSn);
            if (null != list && !list.isEmpty()){ 
                mongoTemplate.insertBatcch(list);
            }
        } finally { 
            //删除缓存，释放空间
            jedisClient.delete(batch);
        }
    }

漂亮，再次测试了下，1万条数据采用了17秒左右。Nice！优化结果大大的超出了预期，我能干的不止如此！我完全可以一次导入10000甚至20000条数据了啊！当然，数据量再大点可能就扛不住了，也不可能无穷大，我还是做个限制，这个限制极限在哪呢？不好控制啊！为何不做成配置呢！

public Response saveOrder(OrderRequest request) { 
       //list.getSize()=10000
       List<Device> list = request.getData;
       //configLimit 使用配置文件，或者配置在DB中
       if (null != list && list.size() > configLimit) { 
           return ResponseUtils.fail("导入设备数量超过限制！");
       }
       ...

至此，本次任务圆满完成！