SpringBoot高效批量插入百万数据

2024-05-01 1214阅读

（图片来源网络，侵删）

前言：我相信很多小伙伴和我一样在初学的时候，面对几万几十万数据插入，不知道如何下手，面对百万级别数据时更是不知所措，我们大部分初学者，很多人都喜欢for循环插入数据，或者是开启多个线程，然后分批使用for循环插入，当我们需要将大量数据存储到数据库中时，传统的逐条插入方式显然效率低下，并且容易导致性能瓶颈。而批量插入是一种更加高效的方式，可以大幅提高数据的插入速度，特别是在数据量较大的情况下。本文将介绍如何使用 Spring Boot 实现高效批量插入百万数据，以解决传统逐条插入方式存在的性能问题。我们将使用不同的插入方式来比较。

1.抛出问题

传统的单条插入存在什么问题：

性能问题：如果循环插入的数据量比较大，每次插入都需要与数据库建立连接、执行插入操作，这将导致频繁的网络通信和数据库操作，性能会受到影响。可以考虑批量插入数据来提高性能，例如使用 JDBC 的批处理功能或使用框架提供的批处理方法。
事务问题：默认情况下，Spring Boot 的事务管理是基于注解的，每次循环插入数据都会开启一个新的事务，这可能导致事务管理的开销过大。可以考虑将整个循环插入数据放在一个事务中，或者使用编程式事务管理来控制事务的粒度。
数据库连接问题：在循环过程中，如果每次都重新获取数据库连接，可能会导致连接资源的浪费和性能下降。可以考虑使用连接池技术来管理数据库连接，确保连接的复用和高效利用。
异常处理问题：在循环插入数据时，可能会出现插入失败、异常等情况。需要适当处理异常，例如记录错误日志、回滚事务等，以确保数据的完整性和一致性。

2.前期准备工作

框架：springboot+mybatis plus +mysql

准备工作

创建一个简单的springboot项目，pom依赖

 
              
                  org.springframework.boot
                  spring-boot-starter-web
              
              
                  com.baomidou
                  mybatis-plus-boot-starter
                  3.4.2
              
              
                  mysql
                  mysql-connector-java
                  8.0.33
              
      
              
                  org.projectlombok
                  lombok
                  true
              
              
                  cn.hutool
                  hutool-all
                  5.8.15
              
              
                  org.springframework.boot
                  spring-boot-starter-test
                  test

创建测试需要使用的表

CREATE TABLE `student` (
        `id` int NOT NULL AUTO_INCREMENT,
        `name` varchar(255) DEFAULT NULL,
        `age` int DEFAULT NULL,
        `addr` varchar(255) DEFAULT NULL,
        `addr_Num` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci DEFAULT NULL,
        PRIMARY KEY (`id`)
      ) ENGINE=InnoDB AUTO_INCREMENT=8497107 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

修改application.yml配置

 server:
        port: 8090
      spring:
        datasource:
          url:  jdbc:mysql://localhost:3306/boot_study?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&rewriteBatchedStatements=true
          username: root
          password: 123456
          driver-class-name: com.mysql.cj.jdbc.Driver

创建实体类StudentDO

注意：在实际业务中，我们应该明确去定义controller service 层的数据模型，数据传输XxxDTO、数据表实体映射XxxDO、返回给前台数据实体XxxVO，这些模型数据都需要根据实际情况在Service实现类和Controller层转换

这里为了演示方便就不按规范定义

@TableName(value = "student")
      @Data
      public class StudentDO {
          /**  主键  type:自增 */
          @TableId(type = IdType.AUTO)
          private Integer id;
      
          /**  名字 */
          private String name;
      
          /**  年龄 */
          private Integer age;
      
          /**  地址 */
          private String addr;
      
          /**  地址号  @TableField：与表字段映射 */
          @TableField(value = "addr_num")
          private String addrNum;
      
          public StudentDO(String name, int age, String addr, String addrNum) {
              this.name = name;
              this.age = age;
              this.addr = addr;
              this.addrNum = addrNum;
          }
      }

controller定义

@RestController
      @RequestMapping("/student")
      public class StudentController {
          @Autowired
          private StudentMapper studentMapper;
          @Autowired
          private StudentService studentService;
          @Autowired
          private SqlSessionFactory sqlSessionFactory;
      
          @Autowired
          private ThreadPoolTaskExecutor taskExecutor;
      
          @Autowired
          private PlatformTransactionManager transactionManager;
          
      
      }

service和impl定义

   public interface StudentService extends IService {
      }
      //实现定义
      @Service
      public class StudentServiceImpl extends ServiceImpl implements StudentService {
      }

Mapper定义

public interface StudentMapper extends BaseMapper {
       @Insert("" +
                  "insert into student (name, age, addr, addr_num) values " +
                  " " +
                  "(#{item.name}, #{item.age},#{item.addr}, #{item.addrNum}) " +
                  " " +
                  "")
          public int insertSplice(@Param("studentDOList") List studentDOList);
      }

3.测试示例演示

模拟100万条数据不同方式插入

for循环单条插入（不建议）
这里100万数据大概要20分钟以上，所以以10万条数据类推
10万条数据总耗时348.864秒

      @GetMapping("/for")
           public void insertForData () {
               long start = System.currentTimeMillis();
       
               for (int i = 0; i  
结果：实际上不知道等了多久很慢很慢，总体时间差不多半个多小时，因为这里的for循环进行单条插入时，每次都是在获取连接(Connection)、释放连接和资源关闭等操作上，（如果数据量大的情况下）极其消耗资源，导致时间长。
2. xml拼接foreach sql插入（大量数据不建议）
10万条数据插入数据耗费时间：3.554秒 
 @GetMapping("/sql")
           public void insertSqlData () {
               long start = System.currentTimeMillis();
               ArrayList arrayList = new ArrayList();
               for (int i = 0; i  
结果：我们在Mapper里面是要insert注解拼接，拼接结果就是将所有的数据集成在一条SQL语句的value值上，其由于提交到服务器上的insert语句少了，相就不需要每次获取连接(Connection)、释放连接和资源关闭，网络负载少了，插入的性能有了提高。但是在数据量大的情况下可能会出现内存溢出、解析SQL语句耗时等情况。
3. mybatis-plus 批量插入saveBatch（推荐）
10万条数据插入数据耗费时间：2.481秒,在数据量不大的情况下和上面差不多
50万条数据插入数据耗费时间：12.473秒 
   @GetMapping("/batch")
           public void insertSaveBatchData () {
               long start = System.currentTimeMillis();
               ArrayList arrayList = new ArrayList();
               for (int i = 0; i  
结果：使用MyBatis-Plus实现IService接口中批处理saveBatch()方法，可以很明显的看到性能有了提升，我们可以查看一下源码，它的底层实现原理利用分片处理（batchSize = 1000） + 分批提交事务的操作，来提高插入性能，并没有在连接上消耗性能，MySQLJDBC驱动默认情况下忽略saveBatch()方法中的executeBatch()语句，将需要批量处理的一组SQL语句进行拆散，执行时一条一条给MySQL数据库，造成实际上是分片插入，即与单条插入方式相比，有提高，但是性能未能得到实质性的提高。 
手动开启批处理模式+批量插入手动提交（推荐）10万条数据插入数据耗费时间：2.481秒,
50万条数据插入数据耗费时间：13.436秒
和上面相比不管是大数据量还是小数据量两者都是差不多
@GetMapping("/forSaveBatch")
           public void insertforSaveBatchData () {
               //创建批量插入SqlSession
               SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false);
               StudentMapper studentMapper = sqlSession.getMapper(StudentMapper.class);
               long start = System.currentTimeMillis();
               ArrayList arrayList = new ArrayList();
               for (int i = 0; i  
结果：手动开启批处理，手动处理关闭自动提交事务，共用同一个SqlSession之后，for循环单条插入的性能得到实质性的提高；由于同一个SqlSession省去对资源相关操作的耗能、减少对事务处理的时间等，从而极大程度上提高执行效率。
5. ThreadPoolTaskExecutor分割多线程插入（大数据量强烈推荐）
50万条数据插入数据耗费时间：3。536秒，插入速度直接是前面的4倍，是不是很疑惑这样就快了这么多?
原理：多线程批量插入的过程，首先定义了一个线程池（ThreadPoolTaskExecutor），用于管理线程的生命周期和执行任务。然后，我们将要插入的数据列表按照指定的批次大小分割成多个子列表，并开启多个线程来执行插入操作，通过 TransactionManager 获取事务管理器，并使用 TransactionDefinition 定义事务属性。然后，在每个线程中，我们通过 transactionManager.getTransaction() 方法获取事务状态，并在插入操作中使用该状态来管理事务。
在插入操作完成后，我们再根据操作结果调用transactionManager.commit()或 transactionManager.rollback() 方法来提交或回滚事务。在每个线程执行完毕后，都会调用 CountDownLatch 的 countDown() 方法，以便主线程等待所有线程都执行完毕后再返回。 
@GetMapping("/threadPoolInsert")
           public void insertThreadPoolBatchData () {
       
       
               ArrayList arrayList = new ArrayList();
               for (int i = 0; i  {
                       DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
                       TransactionStatus status = transactionManager.getTransaction(definition);
                       try {
       
                           studentMapper.insertSplice(StudentDOs);
                           transactionManager.commit(status);
       
                       }catch (Exception e){
                           transactionManager.rollback(status);
                           e.printStackTrace();
       
                       }finally {
                           //执行完后 计数
                           downLatch.countDown();
                       }
                   });
       
               }
               try {
                   //等待
                   downLatch.await();
               } catch (InterruptedException e) {
                   throw new RuntimeException(e);
               }
               long end = System.currentTimeMillis();
               System.out.println("插入数据耗费时间："+(end-start));
           }
 
ThreadPoolTaskExecutor分割异步插入除了上面多线程分割插入，我们也可以使用多线程异步插入其实和上面插入的原理是差不多，下面演示异步插入 
  
修改application.yml增加配置这个参数根据自己的电脑配置合理设置
  async:
           executor:
             thread:
               core_pool_size: 35
               max_pool_size: 35
               queue_capacity: 99999
               name:
                 prefix:  async-testDB-

自定义ThreadPoolTaskExecutor配置

@EnableAsync
         @Configuration
         public class ExecutorConfig {
             @Value("${async.executor.thread.core_pool_size}")
             private int corePoolSize;
             @Value("${async.executor.thread.max_pool_size}")
             private int maxPoolSize;
             @Value("${async.executor.thread.queue_capacity}")
             private int queueCapacity;
             @Value("${async.executor.thread.name.prefix}")
             private String namePrefix;
         
             @Bean(name = "asyncServiceExecutor")
             public Executor asyncServiceExecutor() {
                 ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
                 //设置核心线程数
                 taskExecutor.setCorePoolSize(corePoolSize);
                 //设置最大线程数
                 taskExecutor.setMaxPoolSize(maxPoolSize);
                 //设置队列容量
                 taskExecutor.setQueueCapacity(queueCapacity);
                 //设置线程名前缀
                 taskExecutor.setThreadNamePrefix(namePrefix);
                 //设置拒绝策略
                 // rejection-policy：当pool已经达到max size的时候，如何处理新任务
                 // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
                 taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
                 return taskExecutor;
             }
         
         }

定义异步service和实现类

 //接口服务
         public interface AsyncService {
         
             void executeAsync(List studentDOList, CountDownLatch countDownLatch);
         }
         
         //实现类
         @Service
         public class AsyncServiceImpl extends ServiceImpl implements AsyncService {
             @Autowired
             private StudentService studentService;
             @Async("asyncServiceExecutor")
             @Override
             public void executeAsync(List studentDOList, CountDownLatch countDownLatch) {
                 try{
                     //异步线程要做的事情
                     studentService.saveBatch(studentDOList);
                 }finally {
                     countDownLatch.countDown();// 很关键, 无论上面程序是否异常必须执行countDown,否则await无法释放
                 }
             }
         }

control代码

这里需要注意修改，因为我们在ExecutorConfig配置类里面重新设置了ThreadPoolTaskExecutor

@Autowired

private ThreadPoolTaskExecutor taskExecutor;

改为

@Autowired

private Executor taskExecutor;

@GetMapping("/asyncInsertData")
             public void asyncInsertData() {
                 List studentDOList = getTestData();
                 //测试每100条数据插入开一个线程
                 long start = System.currentTimeMillis();
                 List lists = ListUtil.split(studentDOList, 10000);
                 CountDownLatch countDownLatch = new CountDownLatch(lists.size());
                 for (List listSub:lists) {
                     asyncService.executeAsync(listSub,countDownLatch);
                 }
                 try {
                     countDownLatch.await(); //保证之前的所有的线程都执行完成，才会走下面的；
                     // 这样就可以在下面拿到所有线程执行完的集合结果
                 } catch (Exception e) {
                     e.printStackTrace();
                 }
                 long end = System.currentTimeMillis();
                 System.out.println("插入数据耗费时间："+(end-start));
             }
         
         
             public List getTestData() {
                 ArrayList arrayList = new ArrayList();
                 for (int i = 0; i  
 50万条数据插入数据耗费时间：2.604秒,这里插入和上面差不多因为他们使用的都是多线程插入
 
总结：经过上面的示例演示我们心里已经有谱了，知道什么时候该使用哪一种数据插入方式，针对对不同线程数的测试，发现不是线程数越多越好，具体多少合适，通常的算法：CPU核心数量*2 +2 个线程。 
实际要根据每个人的电脑配置情况设置合适的线程数，可以根据下面这个公式获取： 
 int processNum = Runtime.getRuntime().availableProcessors(); // 返回可用处理器的Java虚拟机的数量
    int corePoolSize = (int) (processNum / (1 - 0.2));
    int maxPoolSize = (int) (processNum / (1 - 0.5));

VPS购买请点击我

免责声明：我们致力于保护作者版权，注重分享，被刊用文章因无法核实真实出处，未能及时与作者取得联系，或有版权异议的，请联系管理员，我们会立即处理! 部分文章是来自自研大数据AI进行生成,内容摘自(百度百科,百度知道,头条百科,中国民法典,刑法,牛津词典,新华词典,汉语词典,国家院校,科普平台)等数据,内容仅供学习参考,不准确地方联系删除处理! 图片声明：本站部分配图来自人工智能系统AI生成,觅知网授权图片,PxHere摄影无版权图库和百度，360，搜狗等多加搜索引擎自动关键词搜索配图，如有侵权的图片，请第一时间联系我们，邮箱：ciyunidc@ciyunshuju.com。本站只作为美观性配图使用,无任何非法侵犯第三方意图,一切解释权归图片著作权方,本站不承担任何责任。如有恶意碰瓷者,必当奉陪到底严惩不贷!

SpringBoot高效批量插入百万数据

相关阅读

怎么把织梦的模板替换?

dedecms怎么调用特定的栏目文档?

怎么抓包一个网页?

wap怎么封装app?

目录[+]