通过一次定时任务的优化复习多线程

大家好，我是后端开发的小扒菜，经常逛掘金看到各位大佬写的文章非常羡慕，技术一般也没啥可写的，正好项目中有一个关于定时任务的需求，用到了多线程，感觉可以写写，路过的朋友可以帮我参谋参谋，有哪里不合适的欢迎交流一下；

需求：用户在交易后都会生成交易凭证信息，让用户在App端输入邮箱，批量下载指定时间内的回单，或者下载单个回单；

背景：

1. 考虑到生成批量的回单压缩文件可能比较慢，且有服务调用，因此做成定时任务，即：用户申请下载回单操作生成一条任务，定时器5分钟扫描一次，将生成好的文件地址通过邮件发送给客户；
2. 历史原因，生成文件的服务，和定时任务的服务是各自独立部署的，因此需要通过服务间调用；

// 第一版
@Scheduled(cron = "0 */5 * * * ?")
public void sendReceiptUrlByEmail() {

    // 查询待处理任务
    List<ReceiptApplyRecordDO> tasks = taskService.queryDownloadTask();
    if(CollUtil.isEmpty(tasks)){
            return;
    }
    for (ReceiptApplyRecordDO task : tasks) {
        // 调用支付中心的回单接口
        ReceiptDownloadUrlDTO param = new ReceiptDownloadUrlDTO();
        // 封装参数

        YiCunTradeVO response = payService.doPost(param, "xxx/mergeserialreceipts");
        if(response.isSuccess()){
                log.info("获取回单地址：{}", response.getData());
                String filePath = (String)response.getData();
                String content = "您的回单下载地址: " + filePath;
                EmailUtil.sendEmail(task.getEmailAddress(),"回单文件地址",content);

                // 更新任务状态
                ReceiptApplyRecordDTO dto = new ReceiptApplyRecordDTO();
                dto.setId(task.getId());
                dto.setStatus(1);
                taskService.updateTaskInfo(dto);
        }else{
                log.info("获取回单地址异常：{}", response.getMsg());
        }
    }
}

第一版中的是根据需求的第一印象写出来代码，考虑发送http请求可能比较耗时，如果有很多任务需要处理，那么势必会耗时比较长，anyway先写出来再优化；

考虑使用多线程优化，首选使用线程池，正好项目中已经配置了线程池可以直接使用，因此将整个请求文件路径操作、发送邮件操作、更新任务状态操作，全部丢给一个线程去执行，通过并行任务来节约时间。

经过测试，30个任务如果是采用版本1的话大概需要50多秒，使用版本2大于是7秒；效果不错；

// 第二版
@Scheduled(cron = "0 */5 * * * ?")
public void sendReceiptUrlByEmail() {
    // 查询待处理任务
    List<ReceiptApplyRecordDO> tasks = taskService.queryDownloadTask();
    if(CollUtil.isEmpty(tasks)){
            return;
    }
    for (ReceiptApplyRecordDO task : tasks) {
        // 调用支付中心的回单接口
        ReceiptDownloadUrlDTO param = new ReceiptDownloadUrlDTO();
        // 封装参数

        threadPoolExecutor.submit(() -> {
            YiCunTradeVO response = payService.doPost(param, "xxx/mergeserialreceipts");
            if(response.isSuccess()){
                String filePath = (String)response.getData();
                String content = "您的回单下载地址: " + filePath;
                EmailUtil.sendEmail(task.getEmailAddress(),"回单文件地址",content);
                // 更新任务状态
                ReceiptApplyRecordDTO dto = new ReceiptApplyRecordDTO();
                dto.setId(task.getId());
                dto.setStatus(1);
                taskService.updateTaskInfo(dto);
        }else{
                log.info("获取回单地址异常：{}", response.getMsg());
            }
        });
    }
}

平常开发中很少使用多线程，想着既然这次用到了，就多尝试一下。对于版本2，每个任务的状态更新都是单独操作的，下一步的优化是使用批量更新，于是有了第三版

// 第三版
@Scheduled(cron = "0 */5 * * * ?")
public void sendReceiptUrlByEmail() {
    // 查询待处理任务
    List<ReceiptApplyRecordDO> tasks = taskService.queryDownloadTask();
    if(CollUtil.isEmpty(tasks)){
            return;
    }
    List<ReceiptApplyRecordDO> waitForUpdate = new ArrayList<>();
    CountDownLatch countDownLatch = new CountDownLatch(tasks.size());
    for (ReceiptApplyRecordDO task : tasks) {
        // 调用支付中心的回单接口
        ReceiptDownloadUrlDTO param = new ReceiptDownloadUrlDTO();
        // 封装参数

        Future<ReceiptApplyRecordDO> result = threadPoolExecutor.submit(() -> {
            YiCunTradeVO response = payService.doPost(param, "xxx/mergeserialreceipts");
            String content = "您的回单下载地址: " + filePath;
            EmailUtil.sendEmail(task.getEmailAddress(),"回单文件地址",content);

            ReceiptApplyRecordDO receiptApplyRecordDO = new ReceiptApplyRecordDO();
            receiptApplyRecordDO.setId(task.getId());
            receiptApplyRecordDO.setStatus(1);
            return receiptApplyRecordDO;
        });
        // 这里是天真的想获取异步执行的结果
        ReceiptApplyRecordDO recordDO = result.get();
        // 这里就是单纯的想用下 countDownLatch，
        countDownLatch.countDown();
        // 收集异步的结果，用来批量更新
        waitForUpdate.add(recordDO);
    }
    countDownLatch.await();
    taskService.updateReceiptTaskBatch(waitForUpdate);
}

这个是失败的版本，同样30个任务的耗时竟然接近版本1了，思路问题，应该是哪里用的不对。通过仔细阅读代码，想起来result.get()是通过阻塞获取结果的，这行代码不会继续往下执行，而是等待直到获取结果，所以才导致每个task的执行都间隔几秒；

继续优化，用一个集合将所有task获取的异步结果收集起来，等任务分发完了统一获取结果，版本4：

// 第四版
@Scheduled(cron = "0 */5 * * * ?")
public void sendReceiptUrlByEmail() {
    try {	
        // 查询待处理任务
        List<ReceiptApplyRecordDO> tasks = taskService.queryDownloadTask();
        if(CollUtil.isEmpty(tasks)){
                return;
        }
        List<ReceiptApplyRecordDO> waitForUpdate = new ArrayList<>();
        List<Future<ReceiptApplyRecordDO>> futureResult = new ArrayList<>();

        for (ReceiptApplyRecordDO task : tasks) {
            // 调用支付中心的回单接口
            ReceiptDownloadUrlDTO param = new ReceiptDownloadUrlDTO();
            // 封装参数

            Future<ReceiptApplyRecordDO> result = threadPoolExecutor.submit(() -> {
                YiCunTradeVO response = payService.doPost(param, "xxx/mergeserialreceipts");

                String content = "您的回单下载地址: " + filePath;
                EmailUtil.sendEmail(task.getEmailAddress(),"回单文件地址",content);

                ReceiptApplyRecordDO receiptApplyRecordDO = new ReceiptApplyRecordDO();
                receiptApplyRecordDO.setId(task.getId());
                receiptApplyRecordDO.setStatus(-1);
                return receiptApplyRecordDO;
            });
            futureResult.add(result);
        }
        for (Future<ReceiptApplyRecordDO> future : futureResult) {
                waitForUpdate.add(future.get());
        }
        taskService.updateReceiptTaskBatch(waitForUpdate);
    } 
}

这个确实实现了异步效果：通过控制台能看到所有的HTTP请求都是一块打印出来的；

总结：

1. 最终版本4和版本3总耗时差不多，如果task这个表被频繁使用，可以考虑使用版本4，毕竟批量更新可以提升数据库性能；否则还是使用版本2，相对简单一些；
2. 多线程真是老大难，主要是用的不多，学了一段时间就忘流，通过这次的迭代优化，又熟悉了一遍线程池的使用；
3. 哪里有写的不合适的地方，希望朋友们指点一下，非常感谢。