批量操作DB

记录一次批量操作数据库，sqlServer服务器参数过多问题。

1.案例引入

对接另一个批发系统B，B需要把订单那些数据弄到系统A中。就是一个批量插入，批量更新的问题。系统保密需要，本文只做示例，具体代码都是模拟的。 (SpringBoot项目)

涉及到的操作类和数据库表

Order_master表【主订单表】

Order_detail表【自订单表，记录主订单里面各个商品品种购买详情，字段数量假设有30个】

Order实体类

OrderMapper

OrderSercice

假设一下批量采购了7500个商品，是不同的种类，

2.old方案

都知道是批量插入数据的吧，构建好数据之后，然后批量插入数据库里面就完事儿了。

//Service
public void insertFun(xxx) {
    ....... 构建数据
    orderDetailMapper.insertOrderList(xxx);
}

<!-- xml -->
<insert id="insertOrderList" parameterType="orderDetail">
        INSERT order_detail (xxxx)VALUES
        <foreach collection="list" item="info" separator="," open="(" close=")">
            #{info.1}, #{info.2}, #{info.3}, #{info.4}, #{info.5}......
        </foreach>
    </insert>

但是这样会有问题。数据库服务器会报错

究其原因： 因为sql server对参数有控制

上面的方法，最终的sql语句为

INSERT order_detail (xxxx)VALUES
(?, ?, ?, ?, ?, ............),
(?, ?, ?, ?, ?, ............),
(?, ?, ?, ?, ?, ............),
.....
(?, ?, ?, ?, ?, ............)

-- 每个对象有30个参数，30 * 7500 = 225000 个参数。。肯定超过了其限制涩。

3.解决方案一

最扯淡的方案：

// service
public void insertFun(xxx) {
    ....... 构建数据
	for ( int i = 0; i < size; ++i ) {
       orderDetailMapper.insert(list.get(i));
    }
}

确实可以解决2100参数的问题。。。。

靠谱一点儿的方案

2100 / 30 = 70 . 我每七十个为一个List，然后批量插入进入不就行了吗？

// service
public void insertFun(xxx) {
    ....... 构建数据
    // 分批，每70个为一批次
	List<List<OrderDetail>> ps = Lists.partition(list, 70);
	for ( List<OrderDetail> t : ps ) {
       orderDetailMapper.insertOrderList(t);
    }
}

这样虽然解决了这个问题，但是，每一次循环伴随SqlSession的打开和关闭，都是一个单独的事务提交，

有7500个商品，7500 / 70≈107。就意味着107SqlSession的打开和关闭，这个是很耗时间的呀。看来还是不能这样搞。

begin
INSERT order_detail (xxxx)VALUES
(?, ?, ?, ?, ?, ............),
......
(?, ?, ?, ?, ?, ............)
commit

begin
INSERT order_detail (xxxx)VALUES
(?, ?, ?, ?, ?, ............),
......
(?, ?, ?, ?, ?, ............)
commit

......
begin
INSERT order_detail (xxxx)VALUES
(?, ?, ?, ?, ?, ............),
......
(?, ?, ?, ?, ?, ............)
commit

那我们就要想办法减少这个连接次数了。

4.saveBatch

然后发现了mybatis-plus的IService接口中的saveBatch。

我们点进去源码查看一下。

// IService.class==========================
public interface IService<T> {
    @Transactional(rollbackFor = {Exception.class}) // 这里有这个注解哦。
    default boolean saveBatch(Collection<T> entityList) {
        return this.saveBatch(entityList, 1000); // 调用了这个saveBatch，可见默认size是1000
    }
    boolean saveBatch(Collection<T> entityList, int batchSize); // 就是这个
}


//ServiceImpl.class==================
public class ServiceImpl<M extends BaseMapper<T>, T> implements IService<T> {
    @Transactional(rollbackFor = {Exception.class})
    public boolean saveBatch(Collection<T> entityList, int batchSize) {
        // SqlMethod.INSERT_ONE是一个枚举
        /*
         INSERT_ONE("insert", "插入一条数据（选择字段插入）", "<script>\nINSERT INTO %s %s VALUES %s\n</script>"),
        */
        String sqlStatement = this.getSqlStatement(SqlMethod.INSERT_ONE);
        // 继续向下执行这个, 第三个参数类型BiConsumer,接收两个参数，不返回东西
        return this.executeBatch(entityList, batchSize, (sqlSession, entity) -> {
            sqlSession.insert(sqlStatement, entity);
        });
    }
    
    // 执行这个
    protected <E> boolean executeBatch(Collection<E> list, int batchSize, BiConsumer<SqlSession, E> consumer) {
        // 调用了SqlHelpr的方法
        return SqlHelper.executeBatch(this.entityClass, this.log, list, batchSize, consumer);
    }
}


//SqlHelper.class====================
public final class SqlHelper {
    public static <E> boolean executeBatch(Class<?> entityClass, Log log, Collection<E> list, int batchSize, BiConsumer<SqlSession, E> consumer) {
        Assert.isFalse(batchSize < 1, "batchSize must not be less than one", new Object[0]);
        // 第三个参数Consumer。
        return !CollectionUtils.isEmpty(list) && executeBatch(entityClass, log, (sqlSession) -> {
            int size = list.size();
            int i = 1;
			// 遍历实体对象的List，
            for(Iterator var6 = list.iterator(); var6.hasNext(); ++i) {
                E element = var6.next();
                consumer.accept(sqlSession, element); // 执行传过来的逻辑sqlSession.insert(sqlStatement, entity);
                if (i % batchSize == 0 || i == size) { // 每一个批次执行一下。(这里是1000)
                    sqlSession.flushStatements();
                }
            }

        });
    }
    
    
     public static boolean executeBatch(Class<?> entityClass, Log log, Consumer<SqlSession> consumer) {
        .......
            
        sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH); // 使用 `ExecutorType.BATCH`表示使用批量操作模式，MyBatis 会将多个 SQL 语句累积起来，而不是每执行一条 SQL 就与数据库交互。
        if (!transaction) {
            log.warn("SqlSession [" + sqlSession + "] was not registered for synchronization because DataSource is not transactional");
        }
        boolean var7;
        try {
            consumer.accept(sqlSession); //---执行传过来的逻辑,就是上面的那个Consumer
            sqlSession.commit(!transaction);
            var7 = true;
        } catch (Throwable var13) {
            .......
        } finally {
            sqlSession.close();
        }

        return var7;
    }
}

从源码中可以发现saveBatch是下面这种情况的

open sqlsession........
begin
sqlSession.insert(sqlStatement, entity);
sqlSession.insert(sqlStatement, entity);
...
sqlSession.insert(sqlStatement, entity); // 1000个
sqlSession.flushStatements();

sqlSession.insert(sqlStatement, entity);
sqlSession.insert(sqlStatement, entity);
...
sqlSession.insert(sqlStatement, entity); // 1000个
sqlSession.flushStatements();
如此循环。。。。。

commit

close sqlsession............

sqlSession.flushStatements()的作用

flushStatements() 方法主要用于批量操作时手动提交累积的 SQL 语句。当使用 ExecutorType.BATCH 时，所有的 SQL 语句会被缓存，直到你调用 flushStatements()，MyBatis 会将缓存中的 SQL 语句一次性提交给数据库执行。这个并不会提交事务哦

我们什么时候可以调用 flushStatements()

• 批量大小控制：如果你的批量数据量比较大，可以控制每批次提交的数据量。通常情况下，合理的批量大小应该根据数据库和应用服务器的配置来调整，太大的批量会导致内存占用过高，太小的批量又不能有效利用批量操作的优势。
• 减少内存压力：如果一次性积累的 SQL 语句过多，会占用大量内存，甚至导致内存超限错误。通过手动调用 flushStatements() 来减少内存的占用，每次提交一个小批次的 SQL 语句。我想这就是默认1000个的原因吧。批量操作时需要分批提交 SQL 语句，减少内存占用、提高性能

这样就可以批量插入数据，同时又避免了2100个参数的问题。

this.saveBatch(整个list)

5.更快的？

结合数据库sqlserver有2100参数限制的问题，还有上面的saveBatch实际上是多条insert语句一起提交到数据库中的，我们可以想到一个更加优秀的解决方案。

• 首先，把一条一条执行insert改为批量的
• 同时，要注意每一条sql的参数限制
• 仿照mybatis-plus源码大致思路执行

每行记录有30个字段，那么，2100/30 = 70，也就是说，要想办法将sql变成下面这个样子的

insert into table1 (x,x,x,....x) values
(x,x,x, .......),(x,x,x, .......),(x,x,x, .......)
(x,x,x, .......).....【70个】

那么，mapper.xml里面的东西可以先不变，就如同第2节里面的<foreach>标签拼接

下面给出实现大致细节。数据源配置，和maven依赖就不放了。

首先，mybatis-plus配置类

@Configuration
public class MybatisPlusConfig {
    //SqlSessionFactory 是 MyBatis 会话工厂，负责创建和配置 SqlSession 实例。这样我们就可以拿到sqlsession了
    @Bean
    public SqlSessionFactory sqlSessionFactory(org.apache.ibatis.session.Configuration configuration) throws Exception {
        MybatisSqlSessionFactoryBean factoryBean = new MybatisSqlSessionFactoryBean();
        factoryBean.setConfiguration(configuration);
        factoryBean.setDataSource(dataSource()); // 注入数据源
        return factoryBean.getObject();
    }

    //SqlSessionTemplate 是 MyBatis 的 SqlSession 的封装，简化了事务的管理。
    @Bean
    public SqlSessionTemplate sqlSessionTemplate(SqlSessionFactory sqlSessionFactory) {
        return new SqlSessionTemplate(sqlSessionFactory); // 创建 SqlSessionTemplate
    }
}

serviceImpl类 手动控制

@Service
public class MyService {
    @Autowired
    private SqlSessionFactory sqlSessionFactory;

    @Autowired
    private OrderDetailMapper orderDetailMapper;

    public void myBatchInsert(List<OrderDetail> details) {
        try (SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH)) {
            OrderDetailMapper mapper = sqlSession.getMapper(OrderDetailMapper.class);

            List<List<OrderDetail>> pats = Lists.partition(details, 65); // 这里调小一丢丢
            int batchCount = 0;
            for (List<OrderDetail> tmp : pats ) {
                mapper.insertOrderList(tmp);
                batchCount++;
                if ( batchCount % 16 == 0 ) { // 16 * 65 = 1040,一个批次插入1040个
                    sqlSession.flushStatements(); // 提交当前批次
                }
            }
            sqlSession.commit(); // 提交事务，持久化到数据库
        } catch(Exception e ) {
            ................
        }
    } 
}

手动挡开着还是爽一些啊，。。

后面我还发现，有这样一个配置jdbc url参数上加rewriteBatchedStatements=true，这个就是把saveBatch由一条一条insert变成了一大条insert，但是一条insert语句太长了的话，sql解析起来估计会更慢了，还是要综合考虑sqlsession的打开与关闭次数，sql数据库解析sql的速度，这三者，才能实现最快最高效的插入数据。 如何综合考虑，得出一个最优方案，估计只有不断测试了，因为每台服务器的配置都不相同。