使用Spring Boot + MyBatis-Plus + ThreadPoolTaskExecutor實(shí)現(xiàn)批量插入百萬級(jí)數(shù)據(jù)
作者：編程疏影
我們充分利用了 Spring Boot 的強(qiáng)大功能、MyBatis-Plus 的便捷操作以及 ThreadPoolTaskExecutor 的高效并發(fā)處理能力，成功實(shí)現(xiàn)了百萬級(jí)數(shù)據(jù)的批量插入。在實(shí)際應(yīng)用中，還可以根據(jù)具體的業(yè)務(wù)需求和性能要求，對(duì)代碼進(jìn)行進(jìn)一步的優(yōu)化和調(diào)整。
在當(dāng)今數(shù)字化高速發(fā)展的時(shí)代，數(shù)據(jù)規(guī)模呈現(xiàn)爆炸式增長(zhǎng)的態(tài)勢(shì)。無論是電子商務(wù)平臺(tái)的海量交易記錄，社交媒體中的海量用戶動(dòng)態(tài)，還是企業(yè)級(jí)應(yīng)用中的復(fù)雜業(yè)務(wù)數(shù)據(jù)，其數(shù)量都極為龐大。在這樣的背景下，如何高效地處理和存儲(chǔ)海量數(shù)據(jù)成為應(yīng)用開發(fā)中至關(guān)重要的挑戰(zhàn)。
批量插入數(shù)據(jù)是眾多系統(tǒng)中頻繁出現(xiàn)的操作場(chǎng)景，特別是在數(shù)據(jù)初始化、數(shù)據(jù)遷移或者高并發(fā)寫入等情境下。然而，當(dāng)面對(duì)百萬級(jí)甚至更龐大的數(shù)據(jù)量時(shí)，傳統(tǒng)的逐個(gè)插入方式往往由于性能瓶頸，導(dǎo)致系統(tǒng)響應(yīng)遲緩，甚至可能出現(xiàn)超時(shí)或崩潰的狀況。
為了有效應(yīng)對(duì)這一嚴(yán)峻挑戰(zhàn)，我們必須采用更為高效的技術(shù)和架構(gòu)策略。Spring Boot 作為強(qiáng)大且成熟的開發(fā)框架，為構(gòu)建穩(wěn)定可靠的應(yīng)用奠定了堅(jiān)實(shí)基礎(chǔ)。MyBatis-Plus 在數(shù)據(jù)操作方面提供了便捷高效的途徑。而 ThreadPoolTaskExecutor 能夠充分發(fā)揮多核 CPU 的優(yōu)勢(shì)，借助并發(fā)處理大幅提升數(shù)據(jù)插入的速度。
通過將這三者有機(jī)結(jié)合，我們能夠構(gòu)建一個(gè)快速、穩(wěn)定且能夠處理百萬級(jí)數(shù)據(jù)批量插入的系統(tǒng)，滿足業(yè)務(wù)對(duì)于數(shù)據(jù)處理的高性能和高可靠性要求。這不僅能夠顯著提升系統(tǒng)的整體性能，還能為用戶帶來更為流暢的使用體驗(yàn)，增強(qiáng)系統(tǒng)在激烈市場(chǎng)競(jìng)爭(zhēng)中的核心競(jìng)爭(zhēng)力。
MySQL DDL 語句
CREATE TABLE `data_table` (
`id` BIGINT(20) PRIMARY KEY AUTO_INCREMENT,
`name` VARCHAR(255),
`description` VARCHAR(255)
);項(xiàng)目創(chuàng)建及依賴配置（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>
<!-- 數(shù)據(jù)庫驅(qū)動(dòng) -->
<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實(shí)體類
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);
}服務(wù)類
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) {
// 分批插入數(shù)據(jù)
int batchSize = 1000; // 每批插入的數(shù)量
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);
}
}
}總結(jié)
通過以上的優(yōu)化和完善，我們充分利用了 Spring Boot 的強(qiáng)大功能、MyBatis-Plus 的便捷操作以及 ThreadPoolTaskExecutor 的高效并發(fā)處理能力，成功實(shí)現(xiàn)了百萬級(jí)數(shù)據(jù)的批量插入。在實(shí)際應(yīng)用中，還可以根據(jù)具體的業(yè)務(wù)需求和性能要求，對(duì)代碼進(jìn)行進(jìn)一步的優(yōu)化和調(diào)整。
責(zé)任編輯：武曉燕
來源：
路條編程
