乐趣区

关于sql:使用Mybatis批量插入大量数据的实践

jiezi

一、前言

最近几天在开发公司业务时，遇到了须要往不数据库中多个表中插入大量数据的一个场景，于是有了这篇文章：

在应用 Mybatis 批量插入数据时的注意事项，以及应用函数式编程对分批次数据插入方法的简略封装。

对于蕴含我在内大部分 Java 后端开发的小伙伴们在平时的 CURD 开发工作中，肯定是免不了应用 Mybatis 这个工具来对数据库进行操作的。

在 SpringBoot 的工程中，引入 Mybatis 后，可应用 mapper 注入的形式来实现增删改查。

比方如果要新增一条数据，那么在 mapper.xml 文件中能够这么写：

        <insert id="testInsert">
        insert into massive_data_insert_test
            (value1,value2)
        values
             (#{value1},#{value2})
    </insert>

而后在 service 层调用 mapper.insertOneItem(insertItem); 即可。

如果要新增多条数据，如果是刚学 Java 的同学可能会这么写：

for(int i = 0; i < insertList.size(); i++){mapper.insertOneItem(insertList.get(i));
}

就是简略的放在一个循环中，屡次调用 mapper 中的数据库新增办法。这种办法编写简略易于了解，在数据量比拟少的时候是不会有什么问题的。

然而一旦数据过多就会有问题了。

实际上每次的 mapper 办法调用都是一次 连贯数据库、预处理（PreparedStatement）、execute（执行 SQL）的过程。

由此发现，如果放在 for 循环中的话，上述过程则会屡次执行，而咱们又分明数据库的连接数是无限的（建设连贯的过程也是很消耗资源的），如果 for 循环的次数过多时，不仅性能会降落还会造成数据库的梗塞甚至程序解体。当然，咱们能够创立或配置数据库连接池（比方 HikariCP、Durid 等）来复用连贯，但这还是对资源的一种节约。

总而言之，如果能有一种办法来一次性把要实现的事件做完就不要分屡次去做。

大部分数据库的新增语句是反对一次插入多条数据的。

insert into table
(value1, value2)
values ('v1','v2'),('v3','v4'),('v1','v5')

Mybatis也给出了一种批量操作数据的办法。应用动静 SQL 语句中的 <foreach> 标签，将帮忙咱们拼接出形似下面的 SQL 语句。

咱们在 mapper.xml 中编写上面的内容：

    <insert id="testMassiveInsert">
        insert into massive_data_insert_test
            (value1,value2)
        values
        <foreach collection="list" item="item" separator=",">
             (#{item.value1},#{item.value2})
        </foreach>
    </insert>

这样咱们就只须要调用一次 mapper 中的办法就能达到下面 for 循环代码的成果。

并且实际上执行的时候也是执行一次 SQL，这个动静SQL 语句的作用就是将传入的参数的内容拼接到插入语句的 SQL 中（预处理技术）。

这种办法很显著要比一开始的 for 循环的实现要好一点了。

二、批量插入数据量达到上万后报错

然而，当咱们用上述的拼接 SQL 的形式进行批量插入时，数据量过大也会呈现问题！

咱们能够先来试验一下批量插入一个四万条数据会怎么。

先来新建一张表作为插入数据的指标表：massive_data_insert_test。

CREATE TABLE "supply"."massive_data_insert_test" ("value1" varchar(255) COLLATE "pg_catalog"."default",
  "value2" varchar(255) COLLATE "pg_catalog"."default"
);

轻易在一个 SpringBoot 的工程中连贯数据库并创立 mapper，编写插入语句（创立工程和扫描mapper 等操作就不在此赘述了）：

上面是 mapper 接口和 mapper.xml 文件（中的 sql 语句）。

TestMapper.java

@Repository
public interface TestMapper.java {void testMassiveInsert(List<HashMap> list);
}

TestMapper.xml

    <insert id="testMassiveInsert">
        insert into massive_data_insert_test
            (value1,value2)
        values
        <foreach collection="list" item="item" separator=",">
             (#{item.value1},#{item.value2})
        </foreach>
    </insert>

测试语句：

@Service
@Slf4j
public class TestService {
        // 批量新增的最大数量
    private static final int maxInsertItemNumPerTime = 500;

    private TestMapper mapper;

    @Autowired
    public TestService(TestMapper mapper) {this.mapper = mapper;}

    public Result testMassiveInsert() {long startTime = System.currentTimeMillis();    // 获取开始工夫
        List<HashMap> list = new LinkedList<>();
          // 组装数据 取得一个长度为 500 * 80 = 40000 的链表
        for (int i = 0; i < maxInsertItemNumPerTime * 80; i++) {HashMap map = new HashMap();
            map.put("value1", "value1" + i);
            map.put("value2", "value2" + i);
            list.add(map);
        }
        // 间接批量插入
        try {mapper.testMassiveInsert(list);
        } catch (RuntimeException e) {log.info("间接批量插入" + list.size() + "失败", e.getMessage());
            throw new RuntimeException(e);
        }
        long endTime = System.currentTimeMillis();    // 获取完结工夫
        return Result.ok().message("程序运行工夫：" + (endTime - startTime) + "ms");
    }
}

当执行下面的间接批量插入时：

间接报出了I/O error，为什么会这样呢？

在前文中提到过 这个动静 SQL 语句的作用就是将传入的参数的内容拼接到插入语句的 SQL 中 ，所以产生这个谬误的起因就是因为要拼接的内容过多，导致SQL 语句过长从而导致了 I/O 的谬误。所以当数据量过大时就会使得拼接过长从而导致程序报错。

并且这个 SQL 的长度不光跟数据量无关，还跟插入语句的插入参数的数量无关。实际上，SQL的长度与二者的乘积呈正相干的线性变动，所以当插入参数过多时更要管制好批量插入数据量的大小。

那么怎么解决呢？最简略的就是分批次插入了，这有点像文章一开始提到的 for 循环中顺次插入的形式，只不过是这回 for 循环中的插入是批量插入了。

因为不是一次性的插入，所以须要加上事务的包裹，从而保障无论哪次插入出错都能回滚。

    @Transactional
    public Result testMassiveInsert() {long startTime = System.currentTimeMillis();    // 获取开始工夫
        List<HashMap> list = new LinkedList<>();
          // 组装数据 取得一个长度为 500 * 80 = 40000 的链表
        for (int i = 0; i < maxInsertItemNumPerTime * 80; i++) {HashMap map = new HashMap();
            map.put("value1", "value1" + i);
            map.put("value2", "value2" + i);
            list.add(map);
        }
        // 分批次的批量插入
        try {if (list.size() > maxInsertItemNumPerTime) {List<List<HashMap>> all = new ArrayList<>();
                int i = 0;
                while (i < list.size()) {List subList = list.subList(i, i + maxInsertItemNumPerTime);
                    i = i + maxInsertItemNumPerTime;
                    all.add(subList);
                }
                all.parallelStream().forEach(o -> mapper.testMassiveInsert(o));
            }
        } catch (RuntimeException e) {log.info("分批次批量插入" + list.size() + "失败", e.getMessage());
            throw new RuntimeException(e);
        }
        long endTime = System.currentTimeMillis();    // 获取完结工夫
        return Result.ok().message("程序运行工夫：" + (endTime - startTime) + "ms");
    }

咱们通过设置一个 maxInsertItemNumPerTime 来管制每次批量插入数据的长度。

三、简略测试

上面我的简略测试（即插入总数量为 4w 条，但设置不同 maxInsertItemNumPerTime 大小时，计算比拟程序运行的耗时）。不过该测试没有进行很多次测试取均匀并又可能网络也有抖动，所以只能算是简略测试。

  • 2000

  • 1000

  • 500

  • 250

最初我抉择了 500 作为分批次插入时每个批次的大小。正如咱们上文所说的一样，即便有数据库连接池提供的连贯复用，然而如果跟数据库的交互多了还是会造成性能的降落，所以这里的 maxInsertItemNumPerTime 的也不是越小越好。

同时，随着 maxInsertItemNumPerTime 的变大，每一次 for 循环中的 SQL 的预处理过程（SQL拼接）耗时会变大，并且这种变大并不是一种线性的，而是往往出现指数型变大(查了一些材料证实了我的这个猜想)，否则的话就不会是 2000 的时候要远远大于 500 了。

大家在理论业务中也须要简略测测来选取一个比拟适合的值，总比没有测试的要好。

四、做点扩大

其实在 Mybatis 的官网文档中是提供了另外一种形式来反对批量插入的。

但因为公司的我的项目中都是用的扫描 Mapper 的形式来操作数据库，加上这种大数据插入场景的确比拟少，所以就没有特意引进上面 Mybatis 提供的形式。

A multiple row insert is a single insert statement that inserts multiple rows into a table. This can be a convenient way to insert a few rows into a table, but it has some limitations:

  1. Since it is a single SQL statement, you could generate quite a lot of prepared statement parameters. For example, suppose you wanted to insert 1000 records into a table, and each record had 5 fields. With a multiple row insert you would generate a SQL statement with 5000 parameters. There are limits to the number of parameters allowed in a JDBC prepared statement – and this kind of insert could easily exceed those limits. If you want to insert many records, you should probably use a JDBC batch insert instead (see below)
  2. The performance of a giant insert statement may be less than you expect. If you have many records to insert, it will almost always be more efficient to use a JDBC batch insert (see below). With a batch insert, the JDBC driver can do some optimization that is not possible with a single large statement
  3. Retrieving generated values with multiple row inserts can be a challenge. MyBatis currently has some limitations related to retrieving generated keys in multiple row inserts that require special considerations (see below)

Nevertheless, there are use cases for a multiple row insert – especially when you just want to insert a few records in a table and don’t need to retrieve generated keys. In those situations, a multiple row insert will be an easy solution.

翻译：

    try (SqlSession session = sqlSessionFactory.openSession()) {GeneratedAlwaysAnnotatedMapper mapper = session.getMapper(GeneratedAlwaysAnnotatedMapper.class);
        List<GeneratedAlwaysRecord> records = getRecordsToInsert(); // not shown
            
        MultiRowInsertStatementProvider<GeneratedAlwaysRecord> multiRowInsert = insertMultiple(records)
                .into(generatedAlways)
                .map(id).toProperty("id")
                .map(firstName).toProperty("firstName")
                .map(lastName).toProperty("lastName")
                .build()
                .render(RenderingStrategies.MYBATIS3);
            
        int rows = mapper.insertMultiple(multiRowInsert);
    }

下面就是文档中的示例代码。

五、进一步优化代码

因为我在公司理论开发中要做的是要往多张表中导入数据，如果依照下面的分批次的写法的话，那我的每一次插入都要有这么一段逻辑：

        // 分批次的批量插入
        try {if (list.size() > maxInsertItemNumPerTime) {List<List<HashMap>> all = new ArrayList<>();
                int i = 0;
                while (i < list.size()) {List subList = list.subList(i, i + maxInsertItemNumPerTime);
                    i = i + maxInsertItemNumPerTime;
                    all.add(subList);
                }
                all.parallelStream().forEach(o -> mapper.testMassiveInsert(o));
            }
        } catch (RuntimeException e) {log.info("分批次批量插入" + list.size() + "失败", e.getMessage());
            throw new RuntimeException(e);
        }

显然这有点反复代码的坏滋味了，看起来就很不好。所以接下来做个简略封装，将这段代码封装到一个办法中，谁插入时就由谁调用。

首先，这段代码须要传入的参数是什么？

  • maxInsertItemNumPerTime 不须要传入，因为显然这是一个常量
  • list 插入的内容链表，须要传入，并且类型不会全都是HashMap，而是不同表对应不同的实体类。须要用到泛型T。
  • mapper中的 testMassiveInsert(HashMap map) 办法，很显著不同表的插入 mapper 必定不是同一个办法，所以这个也须要传入，把一个办法当作参数传入，那么就须要用到 Lambda 表达式和函数式编程。如果你理解过函数式接口，你很天然就会想到像这种只有输出没有输入的函数应该由Consumer 来传入（反之对应Supplier，有输出又有输入时则是Function）。

所以最初形象出的代码应该是这样的：

public <T> void batchSplitInsert(List<T> list, Consumer insertFunc) {List<List<T>> all = new ArrayList<>();
    if (list.size() > maxInsertItemNumPerTime) {
        int i = 0;
        while (i < list.size()) {List subList = list.subList(i, i + maxInsertItemNumPerTime);
            i = i + maxInsertItemNumPerTime;
            all.add(subList);
        }
        all.parallelStream().forEach(insertFunc);
    } else {insertFunc.accept(list);
    }
}

这样子我在做不同表的插入时：

// 待插入数据链表
List<TableDTO> needToInsert = ……；// 进行新增
Consumer<List<TableDTO>> consumer = o -> mapper.insertTable(o);
batchSplitInsert(needToInsert, consumer);

当初整个世界都变优雅了！如果刚入职或者刚学 Java 的小伙伴还对函数式编程不理解，不如先从从简略的 Lambda 语句开始学起，这个真的是 Java8 中超好用的一个个性。

到此为止，本文就算是完结了，大家有啥疑难，请评论留言，咱们互相交换。

如果这篇文章对你有帮忙请多多点赞，感激！如果有大佬有更好的实现形式也请务必不吝赐教！

退出移动版