业务背景

定时任务初始化，调用第三方API 接口获取数据，第三方接口为模糊查询，业务会将需要查询的大量关键词提前，放到TEXT文件中，一行一条数据，项目中是使用定时任务去操作我们的文件，读取获取需要关键字，调用API，获得数据，数据加载到本地DB中。

  1. 业务上传到文件服务器，固定路径中
  2. 触发定时任务，获取文件到本地服务，项目读取文件，加载
  3. 调用API ，获得数据，落库

实际业务实现，出现问题

当需要搜索的关键词比较多，量比较大，这个时候可能由于单线程读取文件，加载比较慢，无法实现快速处理，落库

解决方案：

  1. springboot项目，添加单独线程池，专门用来处理批量任务，与核心业务线程进行区别，保证互不影响，提高安全性
  2. 使用多线程读取本地以及下载好的文件【具体实现下文】文件内容量比较小不建议使用，反而可能造成耗时

项目实践

1. springboot配置类，支持线程数量可配置：application.properties

  1. # 线程池相关 线程池配置
  2. async.film-job.core-pool-size=20
  3. async.film-job.max-pool-size=100
  4. async.film-job.keep-alive-seconds=10
  5. async.film-job.queue-capacity=200
  6. async.film-job.thread-name-prefix=async-Thread-film-service-
  8. # 读取文件开启线程数量
  9. file.thread.num=5

实体类

  1. @Data
  2. @Accessors(chain = true)
  3. public class FileThreadVO<T> {
  4.     private InputStream is;
  5.     private Integer start_line;
  6.     private Integer end_line;
  7.     private List<T> result;
  8. }

2. AsyncFilmServiceConfig线程池配置类：

  2. import lombok.Setter;
  3. import org.springframework.boot.context.properties.ConfigurationProperties;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  6. import org.springframework.scheduling.annotation.EnableAsync;
  7. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  9. import java.util.concurrent.ThreadPoolExecutor;
  11. /**
  12.  *  线程池配置        @Async("asyncOrderService")
  13.  * @EnableAsync开始对异步任务的支持，然后使用@ConfigurationProperties把同类配置信息自动封装成一个实体类
  14.  * @ConfigurationProperties属性prefix表示application.yml配置文件中配置项的前缀，最后结合Lombok的@Setter保证配置文件的值能够注入到该类对应的属性中
  15.  *
  16.  **/
  18. @Setter
  19. @ConfigurationProperties(prefix = "async.film-job")
  20. @EnableAsync
  21. @Configuration
  22. public class AsyncFilmServiceConfig {
  23.     /**
  24.      * 核心线程数（默认线程数）
  25.      */
  26.     private int corePoolSize;
  27.     /**
  28.      * 最大线程数
  29.      */
  30.     private int maxPoolSize;
  31.     /**
  32.      * 允许线程空闲时间（单位：默认为秒）
  33.      */
  34.     private int keepAliveSeconds;
  35.     /**
  36.      * 缓冲队列大小
  37.      */
  38.     private int queueCapacity;
  39.     /**
  40.      * 线程池名前缀
  41.      */
  42.     private String threadNamePrefix;
  44.     @Bean
  45.     public ThreadPoolTaskExecutor asyncFilmService() {
  46.         ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
  47.         threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
  48.         threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);
  49.         threadPoolTaskExecutor.setKeepAliveSeconds(keepAliveSeconds);
  50.         threadPoolTaskExecutor.setQueueCapacity(queueCapacity);
  51.         threadPoolTaskExecutor.setThreadNamePrefix(threadNamePrefix);
  52.         // 线程池对拒绝任务的处理策略
  53.         threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  54.         // 完成任务自动关闭 , 默认为false
  55.         threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
  56.         // 核心线程超时退出，默认为false
  57.         threadPoolTaskExecutor.setAllowCoreThreadTimeOut(true);
  58.         threadPoolTaskExecutor.initialize();
  59.         return threadPoolTaskExecutor;
  60.     }
  61. }

2.2 @Async注解

注意一点：线程池的@Async

  1. @Async注解，它只有一个String类型的value属性，用于指定一个 Bean 的 Name，类型是 Executor 或 TaskExecutor，表示使用这个指定的线程池来执行异步任务：例如 @Async(“asyncFilmService”)

@Async失效：

  1.     ● 如果使用SpringBoot框架必须在启动类中增加@EnableAsync注解<br />                **● 异步方法不能与被调用的异步方法在同一个类中**<br />                ● 异步类没有使用@Component注解（或其他同类注解）导致spring无法扫描到异步类<br />                ● 类中需要使用@Autowired或@Resource等注解自动注入，**不能自己手动new对象**<br />                ● 异步方法使用非public或static修饰

2.3 线程池提交执行线程的原理图

ThreadPoolExecutor

springboot 大文件处理 - 图1

springboot 大文件处理 - 图2

3. 分段读取文件工具类 ReadFileThread

  2. import org.apache.logging.log4j.LogManager;
  3. import org.apache.logging.log4j.Logger;
  5. import java.io.BufferedReader;
  6. import java.io.InputStream;
  7. import java.io.InputStreamReader;
  8. import java.util.ArrayList;
  9. import java.util.List;
  10. import java.util.concurrent.Callable;
  12. /**
  13.  * Description:分段读取文件
  14.  */
  15. public class ReadFileThread implements Callable<List<String>> {
  17.     private static Logger logger = LogManager.getLogger(ReadFileThread.class);
  19.     private Integer start_index;    //文件开始读取行数
  20.     private Integer end_index;      //文件结束读取行数
  21.     private InputStream is;         //输入流
  23.     public ReadFileThread(int start_index, int end_index, InputStream is) {
  24.         this.start_index = start_index;
  25.         this.end_index = end_index;
  26.         this.is = is;
  27.     }
  29.     /**
  30.      * Computes a result, or throws an exception if unable to do so.
  31.      *
  32.      * @return computed result
  33.      * @throws Exception if unable to compute a result
  34.      */
  35.     @Override
  36.     public List<String> call() throws Exception {
  37.         long start = System.currentTimeMillis();
  38.         StringBuilder result = new StringBuilder();
  39.         List<String> resultList = new ArrayList<>();
  40.         BufferedReader reader = new BufferedReader(new InputStreamReader(is, "utf-8"));
  41.         int loc = 1;
  42.         while (loc < start_index) {
  43.             reader.readLine();
  44.             loc++;
  45.         }
  46.         while (loc < end_index) {
  47. //            result.append(reader.readLine()).append("\r\n"); // 读取成string字符串
  48.             resultList.add(reader.readLine().trim());
  49.             loc++;
  50.         }
  51. //        result.append(reader.readLine());
  52.         resultList.add(reader.readLine().trim());
  53. //        String strResult = result.toString();
  54.         reader.close();
  55.         is.close();
  56.         logger.info("线程={} 文件读取完成 总耗时={}毫秒  读取数据={}条",
  57.                 Thread.currentThread().getName(), (System.currentTimeMillis() - start), resultList.size());
  58.         return resultList;
  59.     }
  60. }

4. FileService 服务实现类

  1. import com.zy.website.code.ApiReturnCode;
  2. import com.zy.website.exception.WebsiteBusinessException;
  3. import com.zy.website.model.vo.FileThreadVO;
  4. import com.zy.website.utils.multi.ReadFileThread;
  5. import org.apache.logging.log4j.LogManager;
  6. import org.apache.logging.log4j.Logger;
  7. import org.springframework.beans.factory.annotation.Value;
  8. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  9. import org.springframework.stereotype.Service;
  10. import org.springframework.web.multipart.MultipartFile;
  12. import javax.annotation.Resource;
  13. import java.io.BufferedReader;
  14. import java.io.IOException;
  15. import java.io.InputStream;
  16. import java.io.InputStreamReader;
  17. import java.util.ArrayList;
  18. import java.util.List;
  20. @Service("fileService")
  21. public class FileService{
  22.     //日志
  23.     private static Logger logger = LogManager.getLogger(FileService.class);
  25.     @Value("${file.thread.num}")
  26.     private Integer threadNum; //线程数
  28.     @Resource(name = "asyncFilmService")
  29.     private ThreadPoolTaskExecutor executor;  //线程池
  31.     /**
  32.      * 启用多个线程分段读取文件
  33.      * PS:若文件行数小于线程数会造成线程浪费
  34.      * 适用于读取一行一行的数据报文
  35.      * @return
  36.      */
  37.     public List uploadByThread(MultipartFile file) throws Exception {
  38.         if (file.isEmpty()) {
  39.             return null;
  40.         }
  41.         InputStream is = file.getInputStream();
  42.         List<FileThreadVO> threadVOS = new ArrayList<>(threadNum); //自定义线程实体对象
  43.         //为线程分配读取行数
  44.         Integer lines = getLineNum(is);     //文件总行数
  45.         Integer line;                       //每个线程分配行数
  46.         Integer start_line;                 //线程读取文件开始行数
  47.         Integer end_line;                   //线程读取文件结束行数
  49.         //根据文件行数和线程数计算分配的行数，这里有点繁琐了，待优化
  50.         if (lines < threadNum) {
  51.             for (int i = 1; i <= lines; i++) {
  52.                 FileThreadVO fileThreadVO = new FileThreadVO();
  53.                 start_line = end_line = i;
  54.                 InputStream stream = file.getInputStream();
  56.                 ReadFileThread readFileThread = new ReadFileThread(start_line, end_line, stream);
  57.                 fileThreadVO.setStart_line(start_line);
  58.                 fileThreadVO.setIs(stream);
  59.                 fileThreadVO.setEnd_line(end_line);
  60.                 fileThreadVO.setResult(executor.submit(readFileThread).get());
  61.                 threadVOS.add(fileThreadVO);
  62.             }
  63.         } else {
  64.             for (int i = 1, tempLine = 0; i <= threadNum; i++, tempLine = ++end_line) {
  65.                 InputStream stream = file.getInputStream();
  66.                 FileThreadVO fileThreadVO = new FileThreadVO();
  67.                 Integer var1 = lines / threadNum;
  68.                 Integer var2 = lines % threadNum;
  69.                 line = (i == threadNum) ? (var2 == 0 ? var1 : var1 + var2) : var1;
  70.                 start_line = (i == 1) ? 1 : tempLine;
  71.                 end_line = (i == threadNum) ? lines : start_line + line - 1;
  73.                 ReadFileThread readFileThread = new ReadFileThread(start_line, end_line, stream);
  74.                 fileThreadVO.setStart_line(start_line);
  75.                 fileThreadVO.setIs(stream);
  76.                 fileThreadVO.setEnd_line(end_line);
  77.                 fileThreadVO.setResult(executor.submit(readFileThread).get());
  78.                 threadVOS.add(fileThreadVO);
  79.             }
  80.         }
  81.         List resultCompleteList = new ArrayList<>(); //整合多个线程的读取结果
  82.         threadVOS.forEach(record->{
  83.             List<String> result = record.getResult();
  84.             resultCompleteList.addAll(result);
  85.         });
  87.         boolean isComplete = false;
  88.         if (resultCompleteList != null ) {
  89.             //校验行数 由于本项目使用的是读取行为一个条件 所以只校验行数 也可以校验字节
  90.             int i = resultCompleteList.size() - lines;
  91.             if (i == 0) {
  92.                 isComplete = true;
  93.             }
  94.         }
  95.         if (!isComplete) {
  96.             logger.error(">>>>>====uploadByThread====>>>>>>文件完整性校验失败！");
  97.             throw new WebsiteBusinessException("The file is incomplete！", ApiReturnCode.HTTP_ERROR.getCode());//自定义异常以及错误码
  98.         } else {
  99.             return resultCompleteList;
  100.         }
  101.     }
  103.     /**
  104.      * 获取文件行数
  105.      * @param is
  106.      * @return
  107.      * @throws IOException
  108.      */
  109.     public int getLineNum(InputStream is) throws IOException {
  110.         int line = 0;
  111.         BufferedReader reader = new BufferedReader(new InputStreamReader(is));
  112.         while (reader.readLine() != null) {
  113.             line++;
  114.         }
  115.         reader.close();
  116.         is.close();
  117.         return line;
  118.     }
  119. }

5. 方法中具体使用

springboot 大文件处理 - 图3

该方法只有对文件，利用线程池多线程的读取并未写入，主要业务暂不需要，后续在更。。。