基于java线程池和EasyExcel实现数据异步导入

基于java线程池和EasyExcel实现数据异步导入

2.代码实现

2.1 controller层

1.     @PostMapping("import")
2.     public void importExcel(MultipartFile file) throws IOException {
3.         importService.importExcelAsync(file);
4.     }

复制代码

2.2 service层

1. @Resource
2. private SalariesListener salariesListener;
4. private ExecutorService executorService = Executors.newFixedThreadPool(20);
6. public void importExcelAsync(MultipartFile file) {
7.     // 开20个线程分别处理20个sheet
8.     List<Callable<Object>> tasks = new ArrayList<>();
9.     for (int i = 0; i < 20; i++) {
10.         int num = i;
11.         tasks.add(() -> {
12.             EasyExcel.read(file.getInputStream(), Salaries.class, salariesListener)
13.                     .sheet(num).doRead();
14.             return null;
15.         });
16.     }
18.     try {
19.         //等待所有任务完成
20.         executorService.invokeAll(tasks);
21.     } catch (InterruptedException e) {
22.         throw new RuntimeException(e);
23.     }
25. }

复制代码

2.3实体

1. @Data
2. @TableName("salaries")
3. public class Salaries {
4.     private Integer empNo;
5.     private Integer salary;
6.     private Date fromDate;
7.     private Date toDate;
8. }

复制代码

2.4easyExcel 监听

1. import com.alibaba.excel.context.AnalysisContext;
2. import com.alibaba.excel.read.listener.ReadListener;
3. import com.baomidou.mybatisplus.extension.service.IService;
4. import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
5. import com.woniu.domain.Salaries;
6. import com.woniu.mapper.SalariesMapper;
7. import org.apache.commons.logging.Log;
8. import org.apache.commons.logging.LogFactory;
9. import org.springframework.stereotype.Component;
10. import org.springframework.transaction.annotation.Transactional;
12. import javax.annotation.Resource;
13. import java.util.ArrayList;
14. import java.util.List;
15. import java.util.concurrent.ExecutorService;
16. import java.util.concurrent.Executors;
17. import java.util.concurrent.atomic.AtomicInteger;
19. @Component
20. public class SalariesListener extends ServiceImpl<SalariesMapper, Salaries> implements ReadListener<Salaries>, IService<Salaries> {
22.     private static final Log logger = LogFactory.getLog(SalariesListener.class);
24.     //创建一个线程池，用于异步保存数据
25.     private ExecutorService executorService = Executors.newFixedThreadPool(20);
27.     //创建一个线程安全的list，用于存储读取到的数据，使用ThreadLocal保证线程安全
28.     private ThreadLocal<ArrayList<Salaries>> salariesList = ThreadLocal.withInitial(ArrayList::new);
30.     //用于统计是第几次插入
31.     private static AtomicInteger count = new AtomicInteger(1);
32.    
33.     //设定需要异步批量插入的条数
34.     private static final int batchSize = 10000;
36.     @Resource
37.     private SalariesListener salariesListener;
39.     @Override
40.     @Transactional(rollbackFor = Exception.class)
41.     public void invoke(Salaries data, AnalysisContext context) {
42.         //读取excel每一行的数据,添加到list中
43.         salariesList.get().add(data);
44.         //如果list的数据大于设定需要异步批量插入的条数，则执行异步插入
45.         if (salariesList.get().size() >= batchSize) {
46.             asyncSaveData();
47.         }
48.     }
50.     public void saveData() {
51.         if (!salariesList.get().isEmpty()) {
52.             saveBatch(salariesList.get(), salariesList.get().size());
53.             logger.info("第" + count.getAndAdd(1) + "次插入" + salariesList.get().size() + "条数据");
54.             salariesList.get().clear();
55.         }
56.     }
58.     public void asyncSaveData() {
59.         if (!salariesList.get().isEmpty()) {
60.             ArrayList<Salaries> salaries = (ArrayList<Salaries>) salariesList.get().clone();
61.             executorService.execute(new SaveTask(salaries, salariesListener));
62.             salariesList.get().clear();
63.         }
64.     }
66.     @Override
67.     @Transactional(rollbackFor = Exception.class)
68.     public void doAfterAllAnalysed(AnalysisContext context) {
69.         logger.info("一个Sheet全部处理完");
70.         //考虑每个sheet批量插入数据的条数少于异步插入的条数
71.         asyncSaveData();
72.     }
74.     //创建一个线程类，用于异步保存数据
75.     static class SaveTask implements Runnable {
77.         private List<Salaries> salariesList;
78.         private SalariesListener salariesListener;
80.         public SaveTask(List<Salaries> salariesList, SalariesListener salariesListener) {
81.             this.salariesList = salariesList;
82.             this.salariesListener = salariesListener;
83.         }
85.         @Override
86.         public void run() {
87.             salariesListener.saveBatch(salariesList);
88.             //打印第几次插入，每次插入的数据
89.             logger.info("第" + count.getAndAdd(1) + "次插入" + salariesList.size() + "条数据");
90.         }
91.     }
92. }

复制代码

2.5 建表语句

1. CREATE TABLE `salaries` (
2.   `emp_no` int(11) DEFAULT NULL COMMENT '员工号',
3.   `salary` int(11) DEFAULT NULL,
4.   `from_date` datetime DEFAULT NULL,
5.   `to_date` datetime DEFAULT NULL
6. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

复制代码

1. spring:
2.   servlet:
3.     multipart:
4.       max-request-size: 30MB
5.       max-file-size: 1024MB
6.   datasource:
7.     username: root
8.     password: root
9.     url: jdbc:mysql://127.0.0.1:3306/llp?rewriteBatchedStatements=true&characterEncoding=utf-8&allowPublicKeyRetrieval=true&useSSL=false&serverTimezone=Asia/Shanghai
10.   main:
11.     allow-circular-references: true

复制代码

3.测试验证

可以看到导入95万多条数据，耗时差不多在一份多钟

• 导入开始时间
  

• 导入结束时间
  

• 入库数据
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。