使用Spring Boot + MyBatis-Plus + ThreadPoolTaskExecutor實現批量插入百萬級數據

作者：編程疏影 2024-07-31 09:56:20

我們充分利用了 Spring Boot 的強大功能、MyBatis-Plus 的便捷操作以及 ThreadPoolTaskExecutor 的高效并發處理能力，成功實現了百萬級數據的批量插入。在實際應用中，還可以根據具體的業務需求和性能要求，對代碼進行進一步的優化和調整。

在當今數字化高速發展的時代，數據規模呈現爆炸式增長的態勢。無論是電子商務平臺的海量交易記錄，社交媒體中的海量用戶動態，還是企業級應用中的復雜業務數據，其數量都極為龐大。在這樣的背景下，如何高效地處理和存儲海量數據成為應用開發中至關重要的挑戰。

批量插入數據是眾多系統中頻繁出現的操作場景，特別是在數據初始化、數據遷移或者高并發寫入等情境下。然而，當面對百萬級甚至更龐大的數據量時，傳統的逐個插入方式往往由于性能瓶頸，導致系統響應遲緩，甚至可能出現超時或崩潰的狀況。

為了有效應對這一嚴峻挑戰，我們必須采用更為高效的技術和架構策略。Spring Boot 作為強大且成熟的開發框架，為構建穩定可靠的應用奠定了堅實基礎。MyBatis-Plus 在數據操作方面提供了便捷高效的途徑。而 ThreadPoolTaskExecutor 能夠充分發揮多核 CPU 的優勢，借助并發處理大幅提升數據插入的速度。

通過將這三者有機結合，我們能夠構建一個快速、穩定且能夠處理百萬級數據批量插入的系統，滿足業務對于數據處理的高性能和高可靠性要求。這不僅能夠顯著提升系統的整體性能，還能為用戶帶來更為流暢的使用體驗，增強系統在激烈市場競爭中的核心競爭力。

MySQL DDL 語句

CREATE TABLE `data_table` (
    `id` BIGINT(20) PRIMARY KEY AUTO_INCREMENT,
    `name` VARCHAR(255),
    `description` VARCHAR(255)
);

項目創建及依賴配置（pom.xml）

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.0.0</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.icoderoad</groupId>
    <artifactId>batch-insertion</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>Batch Insertion</name>

    <properties>
        <java.version>17</java.version>
    </properties>

    <dependencies>
        <!-- Spring Boot Web 依賴 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!-- MyBatis-Plus 依賴 -->
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>3.5.3.1</version>
        </dependency>

        <!-- 數據庫驅動 -->
        <dependency>
            <groupId>com.mysql</groupId>
            <artifactId>mysql-connector-j</artifactId>
            <scope>runtime</scope>
        </dependency>

        <!-- 線程池依賴 -->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-tx</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

配置文件（application.yml）

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/db_name?useUnicode=true&characterEncoding=UTF-8&useSSL=false
    username: username
    password: password
    driver-class-name: com.mysql.cj.jdbc.Driver

  task:
    executor:
      core-pool-size: 100
      max-pool-size: 300
      queue-capacity: 99999

實體類

package com.icoderoad.entity;

import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;

@Data
@TableName("data_table")
public class DataEntity {
    private Long id;
    private String name;
    private String description;
}

Mapper 接口

package com.icoderoad.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.icoderoad.entity.DataEntity;

public interface DataMapper extends BaseMapper<DataEntity> {
}

DataService 接口類

package com.icoderoad.service;

import java.util.List;

import com.icoderoad.entity.DataEntity;

public interface DataService {

    void batchInsertData(List<DataEntity> dataList);
}

服務類

package com.icoderoad.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.icoderoad.entity.DataEntity;
import com.icoderoad.mapper.DataMapper;
import com.icoderoad.service.DataService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

@Service
public class DataServiceImpl extends ServiceImpl<DataMapper, DataEntity> implements DataService {

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;

    @Transactional
    public void batchInsertData(List<DataEntity> dataList) {
        // 分批插入數據
        int batchSize = 1000;  // 每批插入的數量

        List<Future<?>> futures = new ArrayList<>();

        for (int i = 0; i < dataList.size(); i += batchSize) {
            List<DataEntity> subList = dataList.subList(i, Math.min(i + batchSize, dataList.size()));
            futures.add(taskExecutor.submit(new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    baseMapper.insertBatchSomeColumn(subList);
                    return null;
                }
            }));
        }

        for (Future<?> future : futures) {
            try {
                future.get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
    }
}

Executor 配置類

package com.icoderoad.config;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
public class ExecutorConfig {

    @Value("${spring.task.executor.core-pool-size}")
    private int corePoolSize;

    @Value("${spring.task.executor.max-pool-size}")
    private int maxPoolSize;

    @Value("${spring.task.executor.queue-capacity}")
    private int queueCapacity;

    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(queueCapacity);
        return executor;
    }
}

控制器類

package com.icoderoad.controller;

import com.icoderoad.entity.DataEntity;
import com.icoderoad.service.DataService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

@RestController
public class DataController {

    @Autowired
    private DataService dataService;

    @PostMapping("/batchInsert")
    public String batchInsert(@RequestBody List<DataEntity> dataEntities) {
        dataService.batchInsertData(dataEntities);
        return "Batch insertion successful";
    }

    public static void main(String[] args) {
        List<DataEntity> dataList = new ArrayList<>();
        Random random = new Random();
        for (int i = 0; i < 1000000; i++) {
            DataEntity dataEntity = new DataEntity();
            dataEntity.setName("Name " + i);
            dataEntity.setDescription("Description " + random.nextInt());
            dataList.add(dataEntity);
        }
    }
}

總結

通過以上的優化和完善，我們充分利用了 Spring Boot 的強大功能、MyBatis-Plus 的便捷操作以及 ThreadPoolTaskExecutor 的高效并發處理能力，成功實現了百萬級數據的批量插入。在實際應用中，還可以根據具體的業務需求和性能要求，對代碼進行進一步的優化和調整。

責任編輯：武曉燕來源：路條編程

Spring 數據優化