工作随笔——Springboot下多线程的使用

一、需求

十万级数据的插入，需要提高插入效率，避免数据堆积。

二、方案

使用多线程异步处理数据插入，多个线程同时插入，提高效率。 1、配置线程池

@Configuration @EnableAsync // 启用异步任务 public class AsyncConfiguration { // 声明一个线程池(并指定线程池的名字) @Bean("taskExecutor") public Executor asyncExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); //核心线程数5：线程池创建时候初始化的线程数 executor.setCorePoolSize(5); //最大线程数5：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程 executor.setMaxPoolSize(5); //缓冲队列500：用来缓冲执行任务的队列 executor.setQueueCapacity(500); //允许线程的空闲时间60秒：当超过了核心线程出之外的线程在空闲时间到达之后会被销毁 executor.setKeepAliveSeconds(60); //线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池 executor.setThreadNamePrefix("DailyAsync-"); executor.initialize(); return executor; } }

2、配置需要异步的方法

@Override @Async("taskExecutor")// 注明线程池名称 public void listInsert(List<HistoryData> list) { log.info("线程" + Thread.currentThread().getId() + "开始执行"); historyDataMapper.insertList(list); log.info("线程" + Thread.currentThread().getId() + "执行结束"); } }

三、问题

在测试类中循环调用此方法，出现问题，子线程还未终止，主线程就已经停止。导致方法并未将数据插入到数据库中。需要在子线程结束前，阻塞主线程。

解决方案：使用CountDownLatch工具类，阻塞主线程。

简单来说，CountDownLatch类是一个计数器，可以设置初始线程数（设置后不能改变），在子线程结束时调用countDown()方法可以使线程数减一，最终为0的时候，调用CountDownLatch的成员方法wait()的线程就会取消BLOKED阻塞状态，进入RUNNABLE从而继续执行。

测试方法

@Test void test11() { try { List<HistoryData> historyDatas = historyDataService.findAll(); int size = historyDatas.size(); int count; if (size%INSERT_NUM==0) { count = size / INSERT_NUM; } else { count = size / INSERT_NUM + 1; } log.info("读入数据成功,共"+size+"条数据"); final CountDownLatch countDownLatch = new CountDownLatch(count); for (int i = 0; i < count; i++) { if (i == 0) { List<HistoryData> list = historyDatas.subList(i, INSERT_NUM); historyDataService.listInsert(list,countDownLatch); } else if ((i + 1) * INSERT_NUM > historyDatas.size()) { List<HistoryData> list = historyDatas.subList((i * INSERT_NUM), historyDatas.size()); historyDataService.listInsert(list,countDownLatch); break; } else { List<HistoryData> list = historyDatas.subList((i * INSERT_NUM), (i + 1) * INSERT_NUM); historyDataService.listInsert(list,countDownLatch); } } countDownLatch.await(); } catch (Exception e) { e.printStackTrace(); } }

service方法

@Override @Async("taskExecutor") public void listInsert(List<HistoryData> list, CountDownLatch countDownLatch) { try { log.info("线程" + Thread.currentThread().getId() + "开始执行"); historyDataMapper.insertList(list); log.info("线程" + Thread.currentThread().getId() + "执行结束"); } finally { //导入完后减1 countDownLatch.countDown(); } }