SpringBoot用多线程批量导入数据库实现方法

作者:愿做无知一猿 时间:2024-01-23 23:03:36 

环境

springboot、mybatisPlus、mysql8

mysql8(部署在1核2G的服务器上,很卡,所以下面的数据条数用5000,太大怕不是要等到花儿都谢了 0.0)

原始的for循环入库

@Service
@Slf4j
public class MoreTestServiceImpl extends ServiceImpl<MoreTestMapper, MoreTestEntity> implements MoreTestService {
   @Override
   @Transactional(rollbackFor = Exception.class)
   public Object doTest() {
       long start = System.currentTimeMillis();
       List<MoreTestEntity> entityList = new ArrayList<>();
       for (int i = 0; i < 5000; i++) {
           MoreTestEntity entity = new MoreTestEntity();
           entity.setId((long) i);
           entity.setA(UUID.randomUUID().toString());
           entity.setB(UUID.randomUUID().toString());
           entity.setC(UUID.randomUUID().toString());
           entity.setD(UUID.randomUUID().toString());
           entity.setE(UUID.randomUUID().toString());
           entity.setF(UUID.randomUUID().toString());
           entity.setG(UUID.randomUUID().toString());
           entity.setH(UUID.randomUUID().toString());
           entity.setI(UUID.randomUUID().toString());
           entity.setJ(UUID.randomUUID().toString());
           entity.setK(UUID.randomUUID().toString());
           entityList.add(entity);
           //在循环中入库
           baseMapper.insert(entity);
       }
       long end = System.currentTimeMillis();
       System.err.println(end - start);
       return end - start;
   }
}

共耗时:180121 ms

批量保存操作

@Service
@Slf4j
public class MoreTestServiceImpl extends ServiceImpl<MoreTestMapper, MoreTestEntity> implements MoreTestService {
   @Override
   @Transactional(rollbackFor = Exception.class)
   public Object doTest() {
       long start = System.currentTimeMillis();
       List<MoreTestEntity> entityList = new ArrayList<>();
       for (int i = 0; i < 5000; i++) {
           MoreTestEntity entity = new MoreTestEntity();
           entity.setId((long) i);
           entity.setA(UUID.randomUUID().toString());
           entity.setB(UUID.randomUUID().toString());
           entity.setC(UUID.randomUUID().toString());
           entity.setD(UUID.randomUUID().toString());
           entity.setE(UUID.randomUUID().toString());
           entity.setF(UUID.randomUUID().toString());
           entity.setG(UUID.randomUUID().toString());
           entity.setH(UUID.randomUUID().toString());
           entity.setI(UUID.randomUUID().toString());
           entity.setJ(UUID.randomUUID().toString());
           entity.setK(UUID.randomUUID().toString());
           entityList.add(entity);
       }
     //mybatisPlus提供的批量保存方法,数字代表每几条数据提交一次事务,默认1000
       saveBatch(entityList, 1000);
       long end = System.currentTimeMillis();
       System.err.println(end - start);
       return end - start;
   }
}

耗时时间:87217ms

在批量插入的基础上使用多线程

@Service
@Slf4j
public class MoreTestServiceImpl extends ServiceImpl<MoreTestMapper, MoreTestEntity> implements MoreTestService {
   @Override
   @Transactional(rollbackFor = Exception.class)
   public Object doTest() throws InterruptedException {
       long start = System.currentTimeMillis();
       //手动创建线程池,注意你 数据库连接池的 允许连接数量,别超过了就行。
       ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
               5,
               5,
               30,
               TimeUnit.SECONDS,
               new LinkedBlockingDeque<>(10),
               //isDaemon 设置线程是否是守护线程,true的话,主线程结束,new的线程就不会继续工作
               new NamedThreadFactory("执行线程", false),
               (r, executor) -> System.out.println("拒绝" + r));
       List<MoreTestEntity> entityList = new ArrayList<>();
       for (int i = 0; i < 5000; i++) {
           MoreTestEntity entity = new MoreTestEntity();
           entity.setId((long) i);
           entity.setA(UUID.randomUUID().toString());
           entity.setB(UUID.randomUUID().toString());
           entity.setC(UUID.randomUUID().toString());
           entity.setD(UUID.randomUUID().toString());
           entity.setE(UUID.randomUUID().toString());
           entity.setF(UUID.randomUUID().toString());
           entity.setG(UUID.randomUUID().toString());
           entity.setH(UUID.randomUUID().toString());
           entity.setI(UUID.randomUUID().toString());
           entity.setJ(UUID.randomUUID().toString());
           entity.setK(UUID.randomUUID().toString());
           entityList.add(entity);
       }
       //拆分list,将其拆分成5份,然后上面线程池创建也是5个核心线程,刚好执行
       List<List<MoreTestEntity>> partition = ListUtils.partition(entityList, 1000);
       //使用CountDownLatch保证所有线程都执行完成
       CountDownLatch latch = new CountDownLatch(5);
       partition.forEach(item -> {
           poolExecutor.execute(() -> {
               saveBatch(item, 1000);
               latch.countDown();
           });
       });
       latch.await();
       // 也可以这么写,设定超时时间
       //latch.await(100,TimeUnit.SECONDS);
       long end = System.currentTimeMillis();
       System.err.println(end - start);
       //关闭线程池
       poolExecutor.shutdown();
       return end - start;
   }
}

耗时时间: 28235

可见时间从180秒,缩短到了28秒,但是@Transactional对于多线程是控制不了所有的事务的。

Spring实现事务的原理是通过ThreadLocal把数据库连接绑定到当前线程中,同一个事务中数据库操作使用同一个jdbc connection,新开启的线程获取不到当前jdbc connection。

如下代码:

partition.forEach(item -> {
           poolExecutor.execute(() -> {
               saveBatch(item, 1000);
               latch.countDown();
               //让每个都报错
               int i = 1/0;
           });
       });

控制台打印:

Exception in thread "执行线程5" java.lang.ArithmeticException: / by zero
    at com.kusch.ares.service.impl.MoreTestServiceImpl.lambda$null$1(MoreTestServiceImpl.java:68)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:750)
Exception in thread "执行线程2" java.lang.ArithmeticException: / by zero
    at com.kusch.ares.service.impl.MoreTestServiceImpl.lambda$null$1(MoreTestServiceImpl.java:68)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:750)
Exception in thread "执行线程4" java.lang.ArithmeticException: / by zero
    at com.kusch.ares.service.impl.MoreTestServiceImpl.lambda$null$1(MoreTestServiceImpl.java:68)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:750)
Exception in thread "执行线程1" java.lang.ArithmeticException: / by zero
    at com.kusch.ares.service.impl.MoreTestServiceImpl.lambda$null$1(MoreTestServiceImpl.java:68)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:750)
Exception in thread "执行线程3" 30179
java.lang.ArithmeticException: / by zero
    at com.kusch.ares.service.impl.MoreTestServiceImpl.lambda$null$1(MoreTestServiceImpl.java:68)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:750)

可见5个线程都报错了,但是去查询数据库,却可以查询到5000条数据,这是不应该出现的情况。

处理多线程入库的事务问题

@Service
@Slf4j
public class MoreTestServiceImpl extends ServiceImpl<MoreTestMapper, MoreTestEntity> implements MoreTestService {
   @Resource
   private DataSourceTransactionManager dataSourceTransactionManager;
   @Resource
   private TransactionDefinition transactionDefinition;
   @Override
   //此处手动管理事务的提交后,这个注解就可以去掉了
   //    @Transactional(rollbackFor = Exception.class)
   public Object doTest() {
       long start = System.currentTimeMillis();
       //手动创建线程池,注意你 数据库连接池的 允许连接数量,别超过了就行。
       ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
               5,
               5,
               30,
               TimeUnit.SECONDS,
               new LinkedBlockingDeque<>(10),
               //isDaemon 设置线程是否是守护线程,true的话,主线程结束,new的线程就不会继续工作
               new NamedThreadFactory("执行线程", false),
               (r, executor) -> System.out.println("拒绝" + r));
       List<MoreTestEntity> entityList = new ArrayList<>();
       for (int i = 0; i < 50; i++) {
           MoreTestEntity entity = new MoreTestEntity();
           entity.setId((long) i);
           entity.setA(UUID.randomUUID().toString());
           entity.setB(UUID.randomUUID().toString());
           entity.setC(UUID.randomUUID().toString());
           entity.setD(UUID.randomUUID().toString());
           entity.setE(UUID.randomUUID().toString());
           entity.setF(UUID.randomUUID().toString());
           entity.setG(UUID.randomUUID().toString());
           entity.setH(UUID.randomUUID().toString());
           entity.setI(UUID.randomUUID().toString());
           entity.setJ(UUID.randomUUID().toString());
           entity.setK(UUID.randomUUID().toString());
           entityList.add(entity);
       }
       //拆分list,将其拆分成5份,然后上面线程池创建也是5个核心线程,刚好执行
       List<List<MoreTestEntity>> partition = ListUtils.partition(entityList, 10);
       //使用CountDownLatch保证所有线程都执行完成
       CountDownLatch sonLatch = new CountDownLatch(5);
       //主线程的 肯定为1
       CountDownLatch mainLatch = new CountDownLatch(1);
       AtomicBoolean hasError = new AtomicBoolean(false);
       partition.forEach(item -> {
           poolExecutor.execute(() -> {
               doSave(item, sonLatch, hasError, mainLatch);
           });
       });
       try {
           //此处应该是用try catch 包裹着主线程的所有业务代码,以此保证主线程中任何一处报错都可以通知子线程
           //这里加一个是为了调试主线程中的数据入库操作
           MoreTestEntity entity = new MoreTestEntity();
           entity.setId((long) 99999);
           entity.setA(UUID.randomUUID().toString());
           entity.setB(UUID.randomUUID().toString());
           entity.setC(UUID.randomUUID().toString());
           entity.setD(UUID.randomUUID().toString());
           entity.setE(UUID.randomUUID().toString());
           entity.setF(UUID.randomUUID().toString());
           entity.setG(UUID.randomUUID().toString());
           entity.setH(UUID.randomUUID().toString());
           entity.setI(UUID.randomUUID().toString());
           entity.setJ(UUID.randomUUID().toString());
           entity.setK(UUID.randomUUID().toString());
           save(entity);
           //主线程报错
           int i = 10 / 0;
           sonLatch.await();
       } catch (InterruptedException e) {
           hasError.set(true);
           e.printStackTrace();
       }
       mainLatch.countDown();
       long end = System.currentTimeMillis();
       System.err.println(end - start);
       //关闭线程池
       if (!poolExecutor.isShutdown()) {
           poolExecutor.shutdown();
       }
       return end - start;
   }
   /**
    * 包装后的子线程的保存代码
    *
    * @param entityList 要保存的集合
    * @param sonLatch   子线程 CountDownLatch
    * @param hasError   是否发生错误
    * @param mainLatch  主线程 CountDownLatch
    */
   private void doSave(List<MoreTestEntity> entityList,
                       CountDownLatch sonLatch,
                       AtomicBoolean hasError,
                       CountDownLatch mainLatch) {
       TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
       try {
           //            //子线程报错
           //            int i = 10 / 0;
           saveBatch(entityList);
       } catch (Throwable throwable) {
           throwable.printStackTrace();
           hasError.set(true);
       } finally {
           //这是必须的,每个子线程走完,要让主线程继续走,然后再回到子线程的每个任务,决定是提交还是回滚
           sonLatch.countDown();
       }
       try {
           //等待主线程的执行结束
           mainLatch.await();
       } catch (InterruptedException e) {
           e.printStackTrace();
           hasError.set(true);
       }
       //事务操作
       if (hasError.get()) {
           dataSourceTransactionManager.rollback(transactionStatus);
       } else {
           dataSourceTransactionManager.commit(transactionStatus);
       }
   }
}

分别放开子线程报错和主线程报错,会发现事务都可以正常回滚,达到了预期的效果。

主要思路就是通过子线程CountDownLatch和主线程CountDownLatch,控制线程好代码的执行顺序即可。

最后补充几点:

  • 上述代码中的countDown()一旦出现不执行的情况那会导致线程堵塞堆积,所以建议给await()增加超时时间

  • 这样操作可能还会出现问题,比如主线程通知子线程可以进行实务操作了,但是各个子线程之间非透明,所以还是有几率存在某个子线程事务回滚失败的情况。

来源:https://blog.csdn.net/qq_38397501/article/details/128853841

标签:SpringBoot,多线程,导入,数据库
0
投稿

猜你喜欢

  • Python3 操作 MySQL 插入一条数据并返回主键 id的实例

    2024-01-21 06:05:16
  • python益智游戏计算汉诺塔问题示例

    2023-02-20 13:02:56
  • OverFlow – 一个秘密武器

    2010-09-25 12:51:00
  • XMLHttpRequest的浏览器兼容代码写法

    2008-09-02 10:46:00
  • Pyinstaller打包工具的使用以及避坑

    2023-11-07 18:50:59
  • 解决Keras使用GPU资源耗尽的问题

    2023-06-26 05:43:51
  • python中map、any、all函数用法分析

    2021-09-02 15:25:45
  • PHP Session条件竞争超详细讲解

    2023-06-03 12:49:00
  • python安装cx_Oracle模块常见问题与解决方法

    2021-04-24 13:00:27
  • MySQL中可为空的字段设置为NULL还是NOT NULL

    2024-01-13 14:38:26
  • python celery分布式任务队列的使用详解

    2021-04-22 18:17:56
  • SQL Server端口更改后的数据库连接方式

    2008-12-29 14:11:00
  • MySQL实现行列转换

    2024-01-18 20:56:17
  • Python趣味爬虫之用Python实现智慧校园一键评教

    2022-02-07 17:53:52
  • 浅谈python下含中文字符串正则表达式的编码问题

    2022-04-08 01:18:35
  • 利用PHP函数计算中英文字符串长度的方法

    2023-10-13 16:35:46
  • pymongo为mongodb数据库添加索引的方法

    2024-01-22 17:51:51
  • mysql下mysqladmin日常管理命令总结(必看篇)

    2024-01-16 23:35:55
  • python 执行终端/控制台命令的例子

    2022-11-16 20:41:29
  • 浅析使用JDBC操作MySQL需要添加Class.forName("com.mysql.jdbc.Driver")

    2024-01-21 23:08:48
  • asp之家 网络编程 m.aspxhome.com