MyBatis-Plus saveBatch 为什么慢：一次百万级批量写库优化实战

摘要

在很多业务系统里，批量写库第一反应往往是直接使用 MyBatis-Plus 的 saveBatch。它足够方便，但当数据量来到 10 万、100 万级时，性能通常并不理想。

本文通过一组实际压测，对比了 4 种常见批量写入方案：

  1. MyBatis-Plus saveBatch 单线程
  2. MyBatis-Plus saveBatch 多线程
  3. mapper.xml 的 foreach values
  4. MyBatis 原生 ExecutorType.BATCH
  5. 纯 JDBC PreparedStatement.addBatch

最终结论是：

  • 单线程 saveBatch 吞吐很低
  • 多线程 saveBatch 有明显提升，但仍有上限
  • foreach values 比 saveBatch 快很多
  • ExecutorType.BATCH 是性能与可维护性的最佳平衡点
  • 纯 JDBC batch 吞吐最高，但业务侵入性也最大

背景

项目中有一类主表数据需要做高吞吐写入。最开始直接使用 MyBatis-Plus 的 saveBatch，功能没有问题，但性能较差。

为了定位瓶颈，我按下面顺序逐步排查：

  1. 调整 MySQL 参数
  2. 校验 JDBC URL 是否开启批处理优化
  3. 排除 SQL 日志输出干扰
  4. 调整批次大小
  5. 从 saveBatch 切到 XML 批量 SQL
  6. 再切到 MyBatis 原生 ExecutorType.BATCH
  7. 最后用纯 JDBC batch 做上限对照

压测时数据库服务器资源占用情况：

  • CPU 约 10%
  • 内存约 30%

这说明数据库本身并没有被打满，瓶颈主要在 Java 侧批处理链路。

配置

1. 测试表

CREATE TABLE `sale_damage_record` (
  `f_id` bigint(50) NOT NULL COMMENT 'ID',
  `project_id` varchar(50) DEFAULT NULL,
  `damage_id` varchar(50) DEFAULT NULL,
  `stage` varchar(50) DEFAULT NULL,
  `stage_name` varchar(500) DEFAULT NULL,
  `start_time` datetime DEFAULT NULL,
  `end_time` datetime DEFAULT NULL,
  `update_time` datetime DEFAULT NULL,
  `flow_id` varchar(50) DEFAULT NULL,
  `flow_task_id` varchar(50) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

如果要支持“按业务列存在则更新，不存在则插入”，还需要唯一索引：

ALTER TABLE sale_damage_record
ADD UNIQUE KEY uk_damage_id (damage_id);

2. MySQL 参数

mysqld.cnf 建议值如下：

[mysqld]
max_allowed_packet=256M
innodb_buffer_pool_size=4G
innodb_log_file_size=512M
innodb_log_buffer_size=128M
innodb_flush_log_at_trx_commit=2
sync_binlog=0

参数含义：

  • max_allowed_packet=256M
    防止批量 SQL 或大包写入时报错
  • innodb_buffer_pool_size=4G
    InnoDB 缓冲池
  • innodb_log_file_size=512M
    redo log 文件大小
  • innodb_log_buffer_size=128M
    redo log buffer
  • innodb_flush_log_at_trx_commit=2
    降低每次提交刷盘成本
  • sync_binlog=0
    降低 binlog 同步开销

3. JDBC URL 参数

spring:
  datasource:
    url: jdbc:mysql://127.0.0.1:3306/test_db?useUnicode=true&characterEncoding=utf-8&serverTimezone=GMT%2B8&useSSL=false&allowMultiQueries=true&rewriteBatchedStatements=true&useServerPrepStmts=true&cachePrepStmts=true&prepStmtCacheSize=500&prepStmtCacheSqlLimit=4096
    username: root
    password: 123456

关键参数说明：

  • rewriteBatchedStatements=true
    让 MySQL 驱动真正优化 batch
  • useServerPrepStmts=true
    开启服务端预编译
  • cachePrepStmts=true
    缓存预编译语句
  • prepStmtCacheSize=500
    预编译缓存数量
  • prepStmtCacheSqlLimit=4096
    可缓存 SQL 长度
  • allowMultiQueries=true
    允许多语句执行

公共实体与造数工具

1. 实体类

package com.example.batch.entity;

import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;

import java.util.Date;

@TableName("sale_damage_record")
public class SaleDamageRecord {

    @TableId("id")
    private Long id;

    @TableField("project_id")
    private String projectId;

    @TableField("damage_id")
    private String damageId;

    @TableField("stage")
    private String stage;

    @TableField("stage_name")
    private String stageName;

    @TableField("start_time")
    private Date startTime;

    @TableField("end_time")
    private Date endTime;

    @TableField("update_time")
    private Date updateTime;

    @TableField("flow_id")
    private String flowId;

    @TableField("flow_task_id")
    private String flowTaskId;

    public String getId() { return id; }
    public void setId(String id) { this.id = id; }

    public String getProjectId() { return projectId; }
    public void setProjectId(String projectId) { this.projectId = projectId; }

    public String getDamageId() { return damageId; }
    public void setDamageId(String damageId) { this.damageId = damageId; }

    public String getStage() { return stage; }
    public void setStage(String stage) { this.stage = stage; }

    public String getStageName() { return stageName; }
    public void setStageName(String stageName) { this.stageName = stageName; }

    public Date getStartTime() { return startTime; }
    public void setStartTime(Date startTime) { this.startTime = startTime; }

    public Date getEndTime() { return endTime; }
    public void setEndTime(Date endTime) { this.endTime = endTime; }

    public Date getUpdateTime() { return updateTime; }
    public void setUpdateTime(Date updateTime) { this.updateTime = updateTime; }

    public String getFlowId() { return flowId; }
    public void setFlowId(String flowId) { this.flowId = flowId; }

    public String getFlowTaskId() { return flowTaskId; }
    public void setFlowTaskId(String flowTaskId) { this.flowTaskId = flowTaskId; }
}

2. 造数工具

package com.example.batch.util;

import com.example.batch.entity.SaleDamageRecord;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;

public class PerfDataBuilder {

    public static SaleDamageRecord buildRecord(int index, String runTag) {
        Date now = new Date();
        ThreadLocalRandom random = ThreadLocalRandom.current();

        SaleDamageRecord record = new SaleDamageRecord();
        record.setId(RandomUtil.uuId());
        record.setProjectId("PROJECT_" + (index % 100));
        record.setDamageId(runTag + "_" + index);
        record.setStage("STAGE_" + ((index % 12) + 1));
        record.setStageName("LOAD_TEST_" + ((index % 12) + 1));
        record.setStartTime(new Date(now.getTime() - random.nextInt(1, 30) * 86400000L));
        record.setEndTime(now);
        record.setUpdateTime(now);
        record.setFlowId("FLOW_" + random.nextInt(1, 100000));
        record.setFlowTaskId("TASK_" + random.nextInt(1, 100000));
        return record;
    }

    public static List<SaleDamageRecord> buildBatch(int startIndex, int batchSize, String runTag) {
        List<SaleDamageRecord> list = new ArrayList<>(batchSize);
        for (int i = 0; i < batchSize; i++) {
            list.add(buildRecord(startIndex + i, runTag));
        }
        return list;
    }

    public static List<SaleDamageRecord> buildAll(int totalCount, String runTag) {
        return buildBatch(1, totalCount, runTag);
    }
}

四种代码方案

方案一：MyBatis-Plus saveBatch 单线程

1. 先说明 saveBatch 第二个参数

service.saveBatch(list, 2500);

这里的第二个参数 2500 表示：

  • 每累计 2500 条
  • MyBatis-Plus 刷一次批
  • 分批提交到数据库

它不是总数，而是批次大小。

2. 可执行代码

package com.example.batch.demo;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.batch.entity.SaleDamageRecord;
import com.example.batch.service.SaleDamageService;
import com.example.batch.util.PerfDataBuilder;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

public class SaveBatchSingleThreadMain {

    private static final int TOTAL_COUNT = 100_000;
    private static final int BATCH_SIZE = 5_000;
    private static final String RUN_TAG = "SB_SINGLE_" + new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());

    public static void main(String[] args) {
        ConfigurableApplicationContext context = new SpringApplicationBuilder(BatchApplication.class)
                .web(WebApplicationType.NONE)
                .run(args);
        try {
            SaleDamageService service = context.getBean(SaleDamageService.class);
            run(service);
        } finally {
            context.close();
        }
    }

    private static void run(SaleDamageService service) {
        long beforeCount = service.count(new QueryWrapper<SaleDamageRecord>()
                .likeRight("damage_id", RUN_TAG));

        long start = System.currentTimeMillis();
        List<SaleDamageRecord> batch = new ArrayList<>(BATCH_SIZE);

        for (int i = 1; i <= TOTAL_COUNT; i++) {
            batch.add(PerfDataBuilder.buildRecord(i, RUN_TAG));
            if (batch.size() == BATCH_SIZE) {
                service.saveBatch(batch, BATCH_SIZE);
                batch.clear();
            }
        }

        if (!batch.isEmpty()) {
            service.saveBatch(batch, BATCH_SIZE);
        }

        long end = System.currentTimeMillis();
        long afterCount = service.count(new QueryWrapper<SaleDamageRecord>()
                .likeRight("damage_id", RUN_TAG));

        System.out.println("insertedCount = " + (afterCount - beforeCount));
        System.out.println("elapsedSeconds = " + ((end - start) / 1000.0d));
    }
}

3. 结果

  • 10万条
  • batchSize = 5000
  • 总耗时约 125 秒

换算吞吐：

  • 100000 / 125 ≈ 800 条/秒

也就是说单线程 saveBatch 的性能大约只有：

  • 800 条/秒

折算 100 万条：

  • 约 1250 秒
  • 约 20 分 50 秒

方案二：MyBatis-Plus saveBatch 多线程

1. 可执行代码

package com.example.batch.demo;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.batch.entity.SaleDamageRecord;
import com.example.batch.service.SaleDamageService;
import com.example.batch.util.PerfDataBuilder;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class SaveBatchThreadPoolMain {

    private static final int TOTAL_COUNT = 1_000_000;
    private static final int BATCH_SIZE = 2_500;
    private static final int CORE_POOL_SIZE = 8;
    private static final int MAX_POOL_SIZE = 16;
    private static final int QUEUE_CAPACITY = 64;
    private static final String RUN_TAG = "SB_MT_" + new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());

    public static void main(String[] args) {
        ConfigurableApplicationContext context = new SpringApplicationBuilder(BatchApplication.class)
                .web(WebApplicationType.NONE)
                .run(args);
        try {
            SaleDamageService service = context.getBean(SaleDamageService.class);
            run(service);
        } finally {
            context.close();
        }
    }

    private static void run(SaleDamageService service) {
        long beforeCount = service.count(new QueryWrapper<SaleDamageRecord>()
                .likeRight("damage_id", RUN_TAG));

        int totalBatches = (TOTAL_COUNT + BATCH_SIZE - 1) / BATCH_SIZE;
        long start = System.currentTimeMillis();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        CountDownLatch latch = new CountDownLatch(totalBatches);
        AtomicInteger finishedRows = new AtomicInteger();

        try {
            for (int batchNo = 0; batchNo < totalBatches; batchNo++) {
                int startIndex = batchNo * BATCH_SIZE + 1;
                int currentBatchSize = Math.min(BATCH_SIZE, TOTAL_COUNT - batchNo * BATCH_SIZE);

                executor.execute(() -> {
                    try {
                        List<SaleDamageRecord> batch =
                                PerfDataBuilder.buildBatch(startIndex, currentBatchSize, RUN_TAG);
                        service.saveBatch(batch, BATCH_SIZE);
                        finishedRows.addAndGet(currentBatchSize);
                    } finally {
                        latch.countDown();
                    }
                });
            }
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        } finally {
            executor.shutdown();
        }

        long end = System.currentTimeMillis();
        long afterCount = service.count(new QueryWrapper<SaleDamageRecord>()
                .likeRight("damage_id", RUN_TAG));

        System.out.println("insertedCount = " + (afterCount - beforeCount));
        System.out.println("finishedRows = " + finishedRows.get());
        System.out.println("elapsedSeconds = " + ((end - start) / 1000.0d));
    }
}

2. 结果

saveBatch 多线程压测结果：

  • 4/8 线程池：10万条 / 26.29秒
  • 6/12 线程池：100万条 / 163.721秒
  • 8/16 线程池：100万条 / 140秒

换算：

  • 8/16 线程池下吞吐约 7140 条/秒

所以正确结论不是“saveBatch = 140s”，而是：

  • 单线程 saveBatch：约 800 条/秒
  • 多线程 saveBatch(8/16)：约 7140 条/秒

方案三：mapper.xml foreach values

1. Mapper

package com.example.batch.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.example.batch.entity.SaleDamageRecord;
import org.apache.ibatis.annotations.Param;

import java.util.List;

public interface SaleDamageMapper extends BaseMapper<SaleDamageRecord> {

    int insertBatchValues(@Param("list") List<SaleDamageRecord> list);

    int insertOrUpdateBatchByDamageId(@Param("list") List<SaleDamageRecord> list);
}

2. 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.example.batch.mapper.SaleDamageMapper">

    <insert id="insertBatchValues">
        insert into sale_damage_record
        (
        id, project_id, damage_id, stage, stage_name,
        start_time, end_time, update_time, flow_id, flow_task_id
        )
        values
        <foreach collection="list" item="item" separator=",">
            (
            #{item.id},
            #{item.projectId},
            #{item.damageId},
            #{item.stage},
            #{item.stageName},
            #{item.startTime},
            #{item.endTime},
            #{item.updateTime},
            #{item.flowId},
            #{item.flowTaskId}
            )
        </foreach>
    </insert>

    <insert id="insertOrUpdateBatchByDamageId">
        insert into sale_damage_record
        (
        id, project_id, damage_id, stage, stage_name,
        start_time, end_time, update_time, flow_id, flow_task_id
        )
        values
        <foreach collection="list" item="item" separator=",">
            (
            #{item.id},
            #{item.projectId},
            #{item.damageId},
            #{item.stage},
            #{item.stageName},
            #{item.startTime},
            #{item.endTime},
            #{item.updateTime},
            #{item.flowId},
            #{item.flowTaskId}
            )
        </foreach>
        on duplicate key update
        project_id = values(project_id),
        stage = values(stage),
        stage_name = values(stage_name),
        start_time = values(start_time),
        end_time = values(end_time),
        update_time = values(update_time),
        flow_id = values(flow_id),
        flow_task_id = values(flow_task_id)
    </insert>

</mapper>

3. 可执行压测代码

package com.example.batch.demo;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.batch.entity.SaleDamageRecord;
import com.example.batch.mapper.SaleDamageMapper;
import com.example.batch.service.SaleDamageService;
import com.example.batch.util.PerfDataBuilder;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MapperXmlThreadPoolMain {

    private static final int TOTAL_COUNT = 1_000_000;
    private static final int BATCH_SIZE = 2_500;
    private static final int CORE_POOL_SIZE = 8;
    private static final int MAX_POOL_SIZE = 16;
    private static final int QUEUE_CAPACITY = 64;
    private static final String RUN_TAG = "XML_MT_" + new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());

    public static void main(String[] args) {
        ConfigurableApplicationContext context = new SpringApplicationBuilder(BatchApplication.class)
                .web(WebApplicationType.NONE)
                .run(args);
        try {
            SaleDamageMapper mapper = context.getBean(SaleDamageMapper.class);
            SaleDamageService service = context.getBean(SaleDamageService.class);
            run(mapper, service);
        } finally {
            context.close();
        }
    }

    private static void run(SaleDamageMapper mapper, SaleDamageService service) {
        long beforeCount = service.count(new QueryWrapper<SaleDamageRecord>()
                .likeRight("damage_id", RUN_TAG));

        int totalBatches = (TOTAL_COUNT + BATCH_SIZE - 1) / BATCH_SIZE;
        long start = System.currentTimeMillis();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        CountDownLatch latch = new CountDownLatch(totalBatches);

        try {
            for (int batchNo = 0; batchNo < totalBatches; batchNo++) {
                int startIndex = batchNo * BATCH_SIZE + 1;
                int currentBatchSize = Math.min(BATCH_SIZE, TOTAL_COUNT - batchNo * BATCH_SIZE);

                executor.execute(() -> {
                    try {
                        List<SaleDamageRecord> batch =
                                PerfDataBuilder.buildBatch(startIndex, currentBatchSize, RUN_TAG);
                        mapper.insertBatchValues(batch);
                    } finally {
                        latch.countDown();
                    }
                });
            }
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        } finally {
            executor.shutdown();
        }

        long end = System.currentTimeMillis();
        long afterCount = service.count(new QueryWrapper<SaleDamageRecord>()
                .likeRight("damage_id", RUN_TAG));

        System.out.println("insertedCount = " + (afterCount - beforeCount));
        System.out.println("elapsedSeconds = " + ((end - start) / 1000.0d));
    }
}

4. 结果

  • 100万条
  • 总耗时 69.148秒

方案四：MyBatis 原生 ExecutorType.BATCH

1. 公共批处理执行器

package com.example.batch.core;

import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;

@Component
public class ParallelMybatisBatchExecutor {

    private final SqlSessionFactory sqlSessionFactory;

    public ParallelMybatisBatchExecutor(SqlSessionFactory sqlSessionFactory) {
        this.sqlSessionFactory = sqlSessionFactory;
    }

    public <T, M> BatchExecuteResult execute(String tenantId,
                                             List<T> dataList,
                                             int batchSize,
                                             int corePoolSize,
                                             int maxPoolSize,
                                             int queueCapacity,
                                             Class<M> mapperClass,
                                             BatchMapperConsumer<M, T> consumer) {

        long start = System.currentTimeMillis();
        List<List<T>> partitions = partition(dataList, batchSize);

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(queueCapacity),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        try {
            List<Future<Integer>> futures = new ArrayList<>(partitions.size());
            for (List<T> partition : partitions) {
                futures.add(executor.submit(buildTask(partition, mapperClass, consumer)));
            }

            int successCount = 0;
            for (Future<Integer> future : futures) {
                try {
                    successCount += future.get();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(e);
                } catch (ExecutionException e) {
                    throw new RuntimeException(e.getCause());
                }
            }

            BatchExecuteResult result = new BatchExecuteResult();
            result.setTotalCount(dataList.size());
            result.setBatchSize(batchSize);
            result.setCorePoolSize(corePoolSize);
            result.setMaxPoolSize(maxPoolSize);
            result.setQueueCapacity(queueCapacity);
            result.setBatchCount(partitions.size());
            result.setSuccessCount(successCount);
            result.setElapsedMs(System.currentTimeMillis() - start);
            return result;
        } finally {
            executor.shutdown();
        }
    }

    private <T, M> Callable<Integer> buildTask(List<T> partition,
                                               Class<M> mapperClass,
                                               BatchMapperConsumer<M, T> consumer) {
        return () -> {
            try (SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false)) {
                M mapper = sqlSession.getMapper(mapperClass);
                for (T data : partition) {
                    consumer.accept(mapper, data);
                }
                sqlSession.flushStatements();
                sqlSession.commit();
                return partition.size();
            }
        };
    }

    private <T> List<List<T>> partition(List<T> dataList, int batchSize) {
        if (dataList.isEmpty()) {
            return Collections.emptyList();
        }
        List<List<T>> partitions = new ArrayList<>((dataList.size() + batchSize - 1) / batchSize);
        for (int i = 0; i < dataList.size(); i += batchSize) {
            partitions.add(dataList.subList(i, Math.min(i + batchSize, dataList.size())));
        }
        return partitions;
    }
}

2. 配套接口与结果类

package com.example.batch.core;

@FunctionalInterface
public interface BatchMapperConsumer<M, T> {
    void accept(M mapper, T data) throws Exception;
}

package com.example.batch.core;

public class BatchExecuteResult {

    private int totalCount;
    private int batchSize;
    private int corePoolSize;
    private int maxPoolSize;
    private int queueCapacity;
    private int batchCount;
    private int successCount;
    private long elapsedMs;

    public int getTotalCount() { return totalCount; }
    public void setTotalCount(int totalCount) { this.totalCount = totalCount; }

    public int getBatchSize() { return batchSize; }
    public void setBatchSize(int batchSize) { this.batchSize = batchSize; }

    public int getCorePoolSize() { return corePoolSize; }
    public void setCorePoolSize(int corePoolSize) { this.corePoolSize = corePoolSize; }

    public int getMaxPoolSize() { return maxPoolSize; }
    public void setMaxPoolSize(int maxPoolSize) { this.maxPoolSize = maxPoolSize; }

    public int getQueueCapacity() { return queueCapacity; }
    public void setQueueCapacity(int queueCapacity) { this.queueCapacity = queueCapacity; }

    public int getBatchCount() { return batchCount; }
    public void setBatchCount(int batchCount) { this.batchCount = batchCount; }

    public int getSuccessCount() { return successCount; }
    public void setSuccessCount(int successCount) { this.successCount = successCount; }

    public long getElapsedMs() { return elapsedMs; }
    public void setElapsedMs(long elapsedMs) { this.elapsedMs = elapsedMs; }
}

3. 可执行压测代码

package com.example.batch.demo;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.batch.core.BatchExecuteResult;
import com.example.batch.core.ParallelMybatisBatchExecutor;
import com.example.batch.entity.SaleDamageRecord;
import com.example.batch.mapper.SaleDamageMapper;
import com.example.batch.service.SaleDamageService;
import com.example.batch.util.PerfDataBuilder;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

public class MybatisBatchThreadPoolMain {

    private static final int TOTAL_COUNT = 1_000_000;
    private static final int BATCH_SIZE = 2_500;
    private static final int CORE_POOL_SIZE = 8;
    private static final int MAX_POOL_SIZE = 16;
    private static final int QUEUE_CAPACITY = 64;
    private static final String RUN_TAG = "MB_MT_" + new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());

    public static void main(String[] args) {
        ConfigurableApplicationContext context = new SpringApplicationBuilder(BatchApplication.class)
                .web(WebApplicationType.NONE)
                .run(args);
        try {
            ParallelMybatisBatchExecutor executor = context.getBean(ParallelMybatisBatchExecutor.class);
            SaleDamageService service = context.getBean(SaleDamageService.class);
            run(executor, service);
        } finally {
            context.close();
        }
    }

    private static void run(ParallelMybatisBatchExecutor executor, SaleDamageService service) {
        long beforeCount = service.count(new QueryWrapper<SaleDamageRecord>()
                .likeRight("damage_id", RUN_TAG));

        List<SaleDamageRecord> entityList = PerfDataBuilder.buildAll(TOTAL_COUNT, RUN_TAG);

        BatchExecuteResult result = executor.execute(
                "tenant-1",
                entityList,
                BATCH_SIZE,
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                QUEUE_CAPACITY,
                SaleDamageMapper.class,
                SaleDamageMapper::insert
        );

        long afterCount = service.count(new QueryWrapper<SaleDamageRecord>()
                .likeRight("damage_id", RUN_TAG));

        System.out.println("insertedCount = " + (afterCount - beforeCount));
        System.out.println("elapsedSeconds = " + (result.getElapsedMs() / 1000.0d));
    }
}

4. 结果

  • 100万条
  • 总耗时 32.616秒

方案五：纯 JDBC batch

1. 可执行代码

package com.example.batch.demo;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.batch.entity.SaleDamageRecord;
import com.example.batch.service.SaleDamageService;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class JdbcBatchThreadPoolMain {

    private static final int TOTAL_COUNT = 1_000_000;
    private static final int BATCH_SIZE = 2_500;
    private static final int CORE_POOL_SIZE = 8;
    private static final int MAX_POOL_SIZE = 16;
    private static final int QUEUE_CAPACITY = 64;
    private static final String RUN_TAG = "JDBC_MT_" + new SimpleDateFormat("yyyyMMddHHmmss").format(new Date());

    private static final String INSERT_SQL =
            "INSERT INTO sale_damage_record " +
            "(id, project_id, damage_id, stage, stage_name, start_time, end_time, update_time, flow_id, flow_task_id) " +
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";

    public static void main(String[] args) {
        ConfigurableApplicationContext context = new SpringApplicationBuilder(BatchApplication.class)
                .web(WebApplicationType.NONE)
                .run(args);
        try {
            DataSource dataSource = context.getBean(DataSource.class);
            SaleDamageService service = context.getBean(SaleDamageService.class);
            run(dataSource, service);
        } finally {
            context.close();
        }
    }

    private static void run(DataSource dataSource, SaleDamageService service) {
        long beforeCount = service.count(new QueryWrapper<SaleDamageRecord>()
                .likeRight("damage_id", RUN_TAG));

        int totalBatches = (TOTAL_COUNT + BATCH_SIZE - 1) / BATCH_SIZE;
        long start = System.currentTimeMillis();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        CountDownLatch latch = new CountDownLatch(totalBatches);

        try {
            for (int batchNo = 0; batchNo < totalBatches; batchNo++) {
                int startIndex = batchNo * BATCH_SIZE + 1;
                int currentBatchSize = Math.min(BATCH_SIZE, TOTAL_COUNT - batchNo * BATCH_SIZE);

                executor.execute(() -> {
                    try {
                        insertBatch(dataSource, startIndex, currentBatchSize);
                    } finally {
                        latch.countDown();
                    }
                });
            }
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        } finally {
            executor.shutdown();
        }

        long end = System.currentTimeMillis();
        long afterCount = service.count(new QueryWrapper<SaleDamageRecord>()
                .likeRight("damage_id", RUN_TAG));

        System.out.println("insertedCount = " + (afterCount - beforeCount));
        System.out.println("elapsedSeconds = " + ((end - start) / 1000.0d));
    }

    private static void insertBatch(DataSource dataSource, int startIndex, int batchSize) {
        try (Connection connection = dataSource.getConnection();
             PreparedStatement ps = connection.prepareStatement(INSERT_SQL)) {
            connection.setAutoCommit(false);
            for (int i = 0; i < batchSize; i++) {
                bind(ps, startIndex + i);
                ps.addBatch();
            }
            ps.executeBatch();
            connection.commit();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private static void bind(PreparedStatement ps, int index) throws Exception {
        Date now = new Date();
        ThreadLocalRandom random = ThreadLocalRandom.current();
        ps.setString(1, UUID.randomUUID().toString().replace("-", ""));
        ps.setString(2, "PROJECT_" + (index % 100));
        ps.setString(3, RUN_TAG + "_" + index);
        ps.setString(4, "STAGE_" + ((index % 12) + 1));
        ps.setString(5, "LOAD_TEST_" + ((index % 12) + 1));
        ps.setTimestamp(6, new Timestamp(now.getTime() - random.nextInt(1, 30) * 86400000L));
        ps.setTimestamp(7, new Timestamp(now.getTime()));
        ps.setTimestamp(8, new Timestamp(now.getTime()));
        ps.setString(9, "FLOW_" + random.nextInt(1, 100000));
        ps.setString(10, "TASK_" + random.nextInt(1, 100000));
    }
}

2. 结果

  • 100万条
  • 总耗时 26秒

结果表格

方案 线程模型 100万数据耗时 吞吐
MyBatis-Plus saveBatch 单线程 约 1250s 约 800 条/秒
MyBatis-Plus saveBatch 8/16 线程池 140s 约 7140 条/秒
mapper.xml foreach values 多线程 69.148s 约 14461 条/秒
MyBatis ExecutorType.BATCH 多线程 32.616s 约 30660 条/秒
纯 JDBC batch 多线程 26s 约 38461 条/秒

资源占用情况

压测过程中，数据库服务器资源占用大致如下：

  • CPU 使用率约 10%
  • 内存使用率约 30%

这说明：

  • 数据库并没有被打满
  • MySQL 本身不是主要瓶颈
  • 性能瓶颈主要在 Java 侧批处理链路

也就是说，这次优化的重点不是继续无脑调 MySQL，而是选择更合适的批处理实现。

结论总结

  1. 单线程 saveBatch 性能非常弱，百万级写入不适合直接使用。
  2. 多线程 saveBatch 虽然能显著提升性能，但框架层开销仍然明显。
  3. mapper.xml foreach values 是一个不错的折中方案，但它依然有“大 SQL 拼接”的成本。
  4. MyBatis 原生 ExecutorType.BATCH 是性能与维护性的最佳平衡点。
  5. 纯 JDBC batch 吞吐最高，适合极限性能场景。
  6. 如果业务需要“根据自定义列进行更新或新增”，可以用：
    insert ... on duplicate key update
    前提是业务列上有唯一索引。
# java # mybatis # mysql # 压力测试
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

四天蒸发近2000亿，宇树科技股价近乎“腰斩”；苹果举办库克告别派对：9月1日正式卸任CEO；Anthropic两款新Claude模型曝光 | 极客头条

avatar AI编程社区
cover

Claude Code反超GitHub Copilot登顶第一、90%程序员已用上Agent，最新AI编码调查报告来了

avatar AI编程社区
cover

如何用好 Claude Code 写后端代码

avatar AI编程社区