Mybatis/MybatisPlus大批量插入操作优化

红太狼 2023-10-09 18:48 97阅读 0赞

## 1 前言 ##

公司的项目存储层框架使用的是mybatis-plus框架,不了解或者没用过的可以去看看官方文档，自带的service,和mapper基本能应付日常的增删改查操作，但是自带的mapper却没有批量插入功能，使用起来很不爽，而且service的saveBatch方法，在处理大批量数据时表现也不尽人意,所以我们进行一些优化

## 2 优化一:JDBC连接URL字符串中需要新增一个参数：rewriteBatchedStatements=true ##

jdbc链接加上：rewriteBatchedStatements=true

url: jdbc:mysql://ip:端口/数据库名?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&autoReconnect=true&rewriteBatchedStatements=true

>  *  MySQL的JDBC连接的url中要加rewriteBatchedStatements参数，并保证5.1.13以上版本的驱动，才能实现高性能的批量插入。
>  *  MySQL JDBC驱动在默认情况下会无视executeBatch()语句，把我们期望批量执行的一组sql语句拆散，一条一条地发给MySQL数据库，批量插入实际上是单条插入，直接造成较低的性能。
>  *  只有把rewriteBatchedStatements参数置为true, 驱动才会帮你批量执行SQL
>  *  这个选项对INSERT/UPDATE/DELETE都有效

PS：这个加上已经可以使你原来的批量插入操作，提升10倍以上

## 3 优化二：使用并行流parallelStream ##

优化之路是无穷无尽的，在网上找了很久，发现可以进一步优化的方案。

> 并行流 :
> 
>  *  把一个内容分成多个数据块，并用不同的线程分 别处理每个数据块的流。
>  *  Java 8 中将并行进行了优化，我们可以很容易的对数据进行并 行操作。Stream API 可以声明性地通过 parallel() 与 sequential() 在并行流与顺序流之间进行切换。

工具类封装：

/**
     * 功能：利用并行流快速插入数据
     *
     * @author wuKeFan
     * @date 2022/11/27
     **/
    public class InsertConsumer {
        
    
        /**
         * 每个长 SQL 插入的行数，可以根据数据库性能调整
         */
        private final static int SIZE = 1000;
    
        /**
         * 如果需要调整并发数目，修改下面方法的第二个参数即可
         */
        static {
        
            System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "4");
        }
    
        /**
         * 插入方法
         *
         * @param list     插入数据集合
         * @param consumer 消费型方法，直接使用 mapper::method 方法引用的方式
         * @param <T>      插入的数据类型
         */
        public static <T> void insertData(List<T> list, Consumer<List<T>> consumer) {
        
            if (list == null || list.size() < 1) {
        
                return;
            }
    
            List<List<T>> streamList = new ArrayList<>();
    
            for (int i = 0; i < list.size(); i += SIZE) {
        
                int j = Math.min((i + SIZE), list.size());
                List<T> subList = list.subList(i, j);
                streamList.add(subList);
            }
            // 并行流使用的并发数是 CPU 核心数，不能局部更改。全局更改影响较大，斟酌
            streamList.parallelStream().forEach(consumer);
        }
    }

使用示例：

@PostMapping("/test/batch")
    public String insertData(@RequestParam int size) {
        
    
        List<BatchTest> list = getList(size);
        // 并行流使用示例1
        InsertConsumer.insertData(list, batchTestService::saveBatch);
        // 并行流使用示例2
        InsertConsumer.insertData(list, batchTestMapper::insertAll);
    
        return "ApiResult.success()";
    }