Spring Boot 集成 Druid 批量插入数据和效率监控配置

Posted ArthurKingYs

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spring Boot 集成 Druid 批量插入数据和效率监控配置相关的知识,希望对你有一定的参考价值。

        最近新的项目写了不少各种 insertBatch 的代码,例如excle导入,批量导入的方式很多,如何选择困扰着大家。下面为大家分析常见的批量插入方法和效率。本文只设计单线程,多线程甚至生产者消费者模式后续补充。

测试环境:

  • SpringBoot 2.5
  • mysql 8
  • JDK 8
  • Druid

搭建测试环境

Druid是Java语言中最好的数据库连接池,并且能够提供强大的监控和扩展功能。

业界把 Druid 和 HikariCP 做对比后,虽说 HikariCP 的性能比 Druid 高,但是因为 Druid 包括很多维度的统计和分析功能,所以这也是大家都选择使用它的原因。

下面来说明如何在 spring Boot 中配置使用Druid

1、添加Maven依赖 (或jar包)
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
            <version>1.0.15</version>
        </dependency>

2、配置数据源相关信息
#datasource
spring.datasource.url=jdbc:mysql://192.168.10.20:3306/test?useUnicode=true&characterEncoding=utf-8&rewriteBatchedStatements=true
spring.datasource.username = root
spring.datasource.password = password
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

#pool Initialization 
spring.datasource.initialSize=5  
spring.datasource.minIdle=5  
spring.datasource.maxActive=20  
#overtime
spring.datasource.maxWait=60000  
spring.datasource.timeBetweenEvictionRunsMillis=60000  
spring.datasource.minEvictableIdleTimeMillis=300000  
spring.datasource.validationQuery=SELECT 1 FROM DUAL  
spring.datasource.testWhileIdle=true  
spring.datasource.testOnBorrow=false  
spring.datasource.testOnReturn=false  
spring.datasource.poolPreparedStatements=true  
spring.datasource.maxPoolPreparedStatementPerConnectionSize=20  
spring.datasource.filters=stat,wall,log4j  
spring.datasource.connectionProperties=druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000
#spring.datasource.useGlobalDataSourceStat=tru

#server
server.port=8081

sql 文件:

drop database IF EXISTS test;
CREATE DATABASE test;
use test;
DROP TABLE IF EXISTS `user`;
CREATE TABLE `user` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT "",
  `age` int(11) DEFAULT 0,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

应用的配置文件:

server:
  port: 8081
spring:
  #数据库连接配置
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://127.0.0.1:3306/test?characterEncoding=utf-8&useSSL=false&allowPublicKeyRetrieval=true&&serverTimezone=UTC&setUnicode=true&characterEncoding=utf8&&nullCatalogMeansCurrent=true&&autoReconnect=true&&allowMultiQueries=true
    username: root
    password: 123456
#mybatis的相关配置
mybatis:
  #mapper配置文件
  mapper-locations: classpath:mapper/*.xml
  type-aliases-package: com.aphysia.spingbootdemo.model
  #开启驼峰命名
  configuration:
    map-underscore-to-camel-case: true
logging:
  level:
    root: error

启动文件,配置了 Mapper 文件扫描的路径:

import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@MapperScan("com.aphysia.springdemo.mapper")
public class SpringdemoApplication 

    public static void main(String[] args) 
        SpringApplication.run(SpringdemoApplication.class, args);
    

Mapper 文件一共准备了几个方法,插入单个对象,删除所有对象,拼接插入多个对象:

import com.aphysia.springdemo.model.User;
import org.apache.ibatis.annotations.Param;

import java.util.List;

public interface UserMapper 

    int insertUser(User user);

    int deleteAllUsers();


    int insertBatch(@Param("users") List<User>users);

Mapper.xml 文件如下:

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<mapper namespace="com.aphysia.springdemo.mapper.UserMapper">
    <insert id="insertUser" parameterType="com.aphysia.springdemo.model.User">
        insert  into user(id,age) values(#id,#age)
    </insert>

    <delete id="deleteAllUsers">
        delete from user where id>0;
    </delete>

    <insert id="insertBatch" parameterType="java.util.List">
        insert into user(id,age) VALUES
        <foreach collection="users" item="model" index="index" separator=",">
            (#model.id, #model.age)
        </foreach>
    </insert>
</mapper>

测试的时候,每次操作我们都删除掉所有的数据,保证测试的客观,不受之前的数据影响。

不同的测试

1. foreach 插入

先获取列表,然后每一条数据都执行一次数据库操作,插入数据:

@SpringBootTest
@MapperScan("com.aphysia.springdemo.mapper")
class SpringdemoApplicationTests 

    @Autowired
    SqlSessionFactory sqlSessionFactory;

    @Resource
    UserMapper userMapper;

    static int num = 100000;

    static int id = 1;

    @Test
    void insertForEachTest() 
        List<User> users = getRandomUsers();
        long start = System.currentTimeMillis();
        for (int i = 0; i < users.size(); i++) 
            userMapper.insertUser(users.get(i));
        
        long end = System.currentTimeMillis();
        System.out.println("time:" + (end - start));
    

2. 拼接sql插入

其实就是用以下的方式插入数据:

INSERT INTO `user` (`id`, `age`) 
VALUES (1, 11),
(2, 12),
(3, 13),
(4, 14),
(5, 15);
@Test
    void insertSplicingTest() 
        List<User> users = getRandomUsers();
        long start = System.currentTimeMillis();
        userMapper.insertBatch(users);
        long end = System.currentTimeMillis();
        System.out.println("time:" + (end - start));
    

3. 使用Batch批量插入

将 MyBatis session 的 executor type 设为 Batch ,使用 sqlSessionFactory 将执行方式置为批量,自动提交置为 false ,全部插入之后,再一次性提交:

@Test
    public void insertBatch()
        SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false);
        UserMapper mapper = sqlSession.getMapper(UserMapper.class);
        List<User> users = getRandomUsers();
        long start = System.currentTimeMillis();
        for(int i=0;i<users.size();i++)
            mapper.insertUser(users.get(i));
        
        sqlSession.commit();
        sqlSession.close();
        long end = System.currentTimeMillis();
        System.out.println("time:" + (end - start));
    

4. 批量处理+分批提交

在批处理的基础上,每1000条数据,先提交一下,也就是分批提交。

@Test
    public void insertBatchForEachTest()
        SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false);
        UserMapper mapper = sqlSession.getMapper(UserMapper.class);
        List<User> users = getRandomUsers();
        long start = System.currentTimeMillis();
        for(int i=0;i<users.size();i++)
            mapper.insertUser(users.get(i));
            if (i % 1000 == 0 || i == num - 1) 
                sqlSession.commit();
                sqlSession.clearCache();
            
        
        sqlSession.close();
        long end = System.currentTimeMillis();
        System.out.println("time:" + (end - start));
    

初次结果,明显不对?

运行上面的代码,我们可以得到下面的结果, for 循环插入的效率确实很差,拼接的 sql 效率相对高一点,看到有些资料说拼接 sql 可能会被 mysql 限制,但是我执行到 1000w 的时候,才看到堆内存溢出。

下面是不正确的结果!!!

插入方式1010010001w10w100w1000w
for循环插入3871150790770026635984太久了...太久了...
拼接sql插入308320392838315624948OutOfMemoryError: 堆内存溢出
批处理392917544251647470666太久了...太久了...
批处理 + 分批提交359893527550270472462太久了...太久了...

拼接sql并没有超过内存

我们看一下 mysql 的限制:

mysql> show VARIABLES like '%max_allowed_packet%';
+---------------------------+------------+
| Variable_name             | Value      |
+---------------------------+------------+
| max_allowed_packet        | 67108864   |
| mysqlx_max_allowed_packet | 67108864   |
| slave_max_allowed_packet  | 1073741824 |
+---------------------------+------------+
3 rows in set (0.12 sec)

这 67108864 足足 600 多M,太大了,怪不得不会报错,那我们去改改一下它吧,改完重新测试:

  1. 首先在启动 mysql 的情况下,进入容器内,也可以直接在 Docker 桌面版直接点 Cli 图标进入:
docker exec -it mysql bash
  1. 进入 /etc/mysql 目录,去修改 my.cnf 文件:
cd /etc/mysql
  1. 先按照 vim ,要不编辑不了文件:
apt-get update
apt-get install vim
  1. 修改 my.cnf
vim my.cnf
  1. 在最后一行添加 max_allowed_packet=20M (按 i 编辑,编辑完按 esc ,输入 :wq 退出)
[mysqld]
pid-file        = /var/run/mysqld/mysqld.pid
socket          = /var/run/mysqld/mysqld.sock
datadir         = /var/lib/mysql
secure-file-priv= NULL
# Disabling symbolic-links is recommended to prevent assorted security risks
symbolic-links=0
 
# Custom config should go here
!includedir /etc/mysql/conf.d/
max_allowed_packet=2M
  1. 退出容器
# exit
  1. 查看 mysql 容器 id
docker ps -a

  1. 重启 mysql
docker restart c178e8998e68

重启成功后查看最大的 max_allowed_pactet ,发现已经修改成功:

mysql> show VARIABLES like '%max_allowed_packet%';
+---------------------------+------------+
| Variable_name             | Value      |
+---------------------------+------------+
| max_allowed_packet        | 2097152    |
| mysqlx_max_allowed_packet | 67108864   |
| slave_max_allowed_packet  | 1073741824 |
+---------------------------+------------+

我们再次执行拼接 sql ,发现 100w 的时候, sql 就达到了 3.6M 左右,超过了我们设置的 2M ,成功的演示抛出了错误:

org.springframework.dao.TransientDataAccessResourceException: 
### Cause: com.mysql.cj.jdbc.exceptions.PacketTooBigException: Packet for query is too large (36,788,583 > 2,097,152). You can change this value on the server by setting the 'max_allowed_packet' variable.
; Packet for query is too large (36,788,583 > 2,097,152). You can change this value on the server by setting the 'max_allowed_packet' variable.; nested exception is com.mysql.cj.jdbc.exceptions.PacketTooBigException: Packet for query is too large (36,788,583 > 2,097,152). You can change this value on the server by setting the 'max_allowed_packet' variable.

批量处理为什么这么慢?

但是,仔细一看就会发现,上面的方式,怎么批处理的时候,并没有展示出优势了,和 for 循环没有什么区别?这是对的么?

这肯定是不对的,从官方文档中,我们可以看到它会批量更新,不会每次去创建预处理语句,理论是更快的。

然后我发现我的一个最重要的问题:数据库连接 URL 地址少了 rewriteBatchedStatements=true

如果我们不写, MySQL JDBC 驱动在默认情况下会忽视 executeBatch() 语句,我们期望批量执行的一组 sql 语句拆散,但是执行的时候是一条一条地发给 MySQL 数据库,实际上是单条插入,直接造成较低的性能。我说怎么性能和循环去插入数据差不多。

只有将 rewriteBatchedStatements 参数置为 true , 数据库驱动才会帮我们批量执行 SQL 。

正确的数据库连接:

jdbc:mysql://127.0.0.1:3306/test?characterEncoding=utf-8&useSSL=false&allowPublicKeyRetrieval=true&&serverTimezone=UTC&setUnicode=true&characterEncoding=utf8&&nullCatalogMeansCurrent=true&&autoReconnect=true&&allowMultiQueries=true&&&rewriteBatchedStatements=true

找到问题之后,我们重新测试批量测试,最终的结果如下:

插入方式1010010001w10w100w1000w
for循环插入3871150790770026635984太久了...太久了...
拼接sql插入308320392838315624948(很可能超过sql长度限制)OutOfMemoryError: 堆内存溢出
批处理(重点)33332336263616388978OutOfMemoryError: 堆内存溢出
批处理 + 分批提交359313394630290718631OutOfMemoryError: 堆内存溢出

从上面的结果来看,确实批处理是要快很多的,当数量级太大的时候,其实都会超过内存溢出的,批处理加上分批提交并没有变快,和批处理差不多,反而变慢了,提交太多次了,拼接 sql 的方案在数量比较少的时候其实和批处理相差不大,最差的方案就是 for 循环插入数据,这真的特别的耗时。 100 条的时候就已经需要 1s 了,不能选择这种方案。

一开始发现批处理比较慢的时候,真的挺怀疑自己,后面发现是有一个参数,有一种拨开云雾的感觉,知道得越多,不知道的越多。

以上是关于Spring Boot 集成 Druid 批量插入数据和效率监控配置的主要内容,如果未能解决你的问题,请参考以下文章

Spring Boot 集成 Druid 批量插入数据和效率监控配置

Spring boot 集成 Druid 数据源

Spring boot 集成 分页,druid,redis

Spring boot 集成 分页,druid,redis

Spring boot 集成 分页,druid,redis

spring boot druid动态多数据源监控集成