本文主要讲述通过Spring Boot + MyBatis做大数据量数据插入的案例和结果
不分批次直接梭哈
MyBatis直接一次性批量插入30万条，代码如下：
@Test
public void testBatchInsertUser() throws IOException {
InputStream resourceAsStream =
Resources.getResourceAsStream("sqlMapConfig.xml");
SqlSessionFactory sqlSessionFactory = new SqlSessionFactoryBuilder().build(resourceAsStream);
SqlSession session = sqlSessionFactory.openSession();
System.out.println("===== 开始插入数据 =====");
long startTime = System.currentTimeMillis();
try {
List<TestImport> entityList = new ArrayList<>();
for(int i =0; i< 300000; i++) {
TestImport testImport = new TestImport();
testImport.setName(RandomUtils.getLetters(6));
testImport.setPhone(RandomUtils.getNumbers(13));
entityList.add(testImport);
}
session.insert("saveBatch", entityList); // 最后插入数据
session.commit();
long spendTime = System.currentTimeMillis()-startTime;
System.out.println("成功插入 30 万条数据,耗时："+spendTime+"毫秒");
} finally {
session.close();
}
}可以看到控制台输出：
超出最大数据包限制了，可以通过调整max_allowed_packet限制来提高可以传输的内容，不过由于30万条数据超出太多，这个不可取，梭哈看来是不行了 既然梭哈不行那我们就一条一条循环着插入行不行呢，这种方法我没有自己亲自尝试30w的数据，但是我平时尝试了一千多条，耗时都要分钟级别的，这种一条一条循环着插入不建议直接使用。
分批插入
1. MyBatis手写Sql分批插入
MyBatis直接手写batchInsert方法的Sql我就不写了，大家可以自己写
@Test
public void testBatchInsertUser() throws IOException {
InputStream resourceAsStream = Resources.getResourceAsStream("sqlMapConfig.xml");
SqlSessionFactory sqlSessionFactory = new SqlSessionFactoryBuilder().build(resourceAsStream);
SqlSession session = sqlSessionFactory.openSession();
System.out.println("===== 开始插入数据 =====");
long startTime = System.currentTimeMillis();
int waitTime = 10;
try {
List<User> entityList = new ArrayList<>();
for (int i = 1; i <= 300000; i++) {
TestImport testImport = new TestImport();
testImport.setName(RandomUtils.getLetters(6));
testImport.setPhone(RandomUtils.getNumbers(13));
entityList.add(testImport);
if (i % 1000 == 0) {
session.insert("batchInsert", entityList);
// 每 1000 条数据提交一次事务
session.commit();
entityList.clear();
// 等待一段时间
Thread.sleep(waitTime * 1000);
}
}
// 最后插入剩余的数据
if(!CollectionUtils.isEmpty(entityList)) {
session.insert("batchInsert", entityList);
session.commit();
}
long spendTime = System.currentTimeMillis()-startTime;
System.out.println("成功插入 30 万条数据,耗时："+spendTime+"毫秒");
} catch (Exception e) {
e.printStackTrace();
} finally {
session.close();
}
}2. MyBatis- Plus分批插入
这种方法可以详见我之前写的文章
解决Mybatis-plus 批量插入太慢的问题，提升插入性能_Crystalqy的博客-CSDN博客
使用多线程分批插入，并且控制事物
使用多线程分批插入，是基于MyBatis- Plus分批插入做了一个并行处理的优化，使用多线程同时处理数据的插入，这里重点是事物的控制
我的事物要求是多个线程同时插入，只有所有的线程都执行成功了，主线程最后才能提交，只要有一个线程失败了，所有的事物都要回滚，为了达到这种目的，可以使用TransactionTemplate和CountDownLatch来实现这个目标。
1. TransactionTemplate + CountDownLatch
以下是一个示例代码，演示了如何在多个线程中同时插入不同的对象，然后在所有线程执行完成后进行事务的提交或回滚：
@Service
public class ConcurrentInsertService {
@Autowired
private TransactionTemplate transactionTemplate;
@Autowired
private YourRepository repository;
public void performConcurrentInsert() throws InterruptedException {
int threadCount = 3;
CountDownLatch latch = new CountDownLatch(threadCount);
AtomicBoolean allSuccessful = new AtomicBoolean(true); // 用于记录所有线程是否成功
for (int i = 0; i < threadCount; i++) {
new Thread(() -> {
try {
transactionTemplate.execute(status -> {
try {
// 执行插入操作，例如：
YourObject object = new YourObject();
// 设置对象属性
repository.save(object);
} catch (Exception e) {
status.setRollbackOnly();
allSuccessful.set(false); // 标记为失败
}
return null;
});
} finally {
latch.countDown();
}
}).start();
}
latch.await(); // 等待所有线程完成
if (allSuccessful.get()) {
// 所有线程都成功，进行事务提交操作
transactionTemplate.execute(status -> {
// 执行一些整体提交操作
return null;
});
} else {
// 至少一个线程失败，进行事务回滚操作
transactionTemplate.execute(status -> {
// 执行一些整体回滚操作
return null;
});
}
}
}
在这个示例中，我使用了AtomicBoolean来标记是否所有线程都成功执行。在等待所有线程完成后，根据allSuccessful的值，你可以决定是提交事务还是回滚事务。请根据你的实际业务需求，适当调整示例代码中的提交和回滚操作。
请注意，transactionTemplate.execute 方法会在事务作用域内执行给定的代码块，如果在代码块中抛出异常，会导致事务回滚。这就是为什么在上面的代码中，我使用了setRollbackOnly 来标记失败的事务。整体的事务提交或回滚操作也是在自己的事务作用域内执行的。
2. TransactionManager + TransactionStatus
这个是本文重点要介绍的方法
下面是我本次的代码：
API层的代码：
@GetMapping(value = "/testImport")
public void testImport() throws InterruptedException {
List<TestImport> entityList = new ArrayList<>();
for(int i =0; i< 10000; i++) {
TestImport testImport = new TestImport();
testImport.setName(RandomUtils.getLetters(6));
testImport.setPhone(RandomUtils.getNumbers(13));
entityList.add(testImport);
}
importService.performConcurrentImport(entityList);
}Service成核心代码：核心代码的注释都已经写的很清楚了
@Service
public class TestImportService {
@Autowired
private PlatformTransactionManager transactionManager; // 事务管理器
@Transactional(propagation = Propagation.REQUIRED, rollbackFor = {Exception.class})
public void performConcurrentImport(List<TestImport> entityList) throws InterruptedException {
int threadCount = 3;
if (entityList.size() < 1000) {
threadCount = 1;
} else if (entityList.size() < 5000) {
threadCount = 2;
}
// 创建多线程处理任务
ExecutorService threadPool = Executors.newFixedThreadPool(threadCount);
AtomicBoolean allsuccessfull = new AtomicBoolean(true); // 用于记录所有线程是否成功
TestImportService importService = SpringContextUtil.getApplicationContext().getBean(TestImportService.class);
List<TransactionStatus> transactionStatuses = Collections.synchronizedList(new ArrayList<TransactionStatus>());
try {
List<FutureTask<TransactionStatus>> tasks = Lists.newArrayList();
List<List<TestImport>> subList = ListUtils.subWithNum(entityList, threadCount);
for (int i = 0; i < threadCount; i++) {
List<TestImport> insertList = subList.get(i);
FutureTask<TransactionStatus> importTask = new FutureTask<>(() -> importService.importTransaction(transactionManager, insertList));
tasks.add(importTask);
}
for (FutureTask<TransactionStatus> futureTask : tasks) {
threadPool.submit(futureTask);
}
try {
for (FutureTask<TransactionStatus> futureTask : tasks) {
transactionStatuses.add(futureTask.get());
}
} catch (Exception e) {
e.printStackTrace();
allsuccessfull.set(false);
} finally {
}
} catch (Exception e) {
e.printStackTrace();
allsuccessfull.set(false);
}
if (!transactionStatuses.isEmpty()) {
if (allsuccessfull.get()) { //全部执行成功，提交事物
transactionStatuses.forEach(s -> transactionManager.commit(s));
} else {// 只要有一个线程执行失败，就回滚事物
transactionStatuses.forEach(s -> transactionManager.rollback(s));
}
}
System.out.println("主线程完成");
}
/**
* 使用这种方式将事务状态都放在同一个事务里面
* @param transactionManager
* @param entityList
* @return
*/
@Transactional(propagation = Propagation.REQUIRED, rollbackFor = {Exception.class})
public TransactionStatus importTransaction(PlatformTransactionManager transactionManager, List<TestImport> entityList ) {
DefaultTransactionDefinition definition = new DefaultTransactionDefinition();// 事务定义接口：事务的一些基础信息，如超时时间、隔离级别、传播属性等
definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); // 事物传播属性，开启新事务，这样会比较安全些。
TransactionStatus status = transactionManager.getTransaction(definition); // 获得事务状态，事务的一些状态信息，如是否是一个新的事务、是否已被标记为回滚
super.saveBatch(0, entityList);
System.out.println("线程名称："+ Thread.currentThread().getName());
return status;
}
}3. TransactionManager和TransactionTemplate的简单介绍
当在Spring应用程序中处理数据库事务时，`TransactionManager`和`TransactionTemplate`是两个重要的组件。它们都用于管理和控制事务的开启、提交、回滚等操作。
1. **TransactionManager：**
`TransactionManager`是Spring框架中用于管理事务的核心接口。它提供了在应用程序中执行事务相关操作的标准方法。具体来说，`TransactionManager`有助于以下几个方面：
- **事务的开启和提交：** `TransactionManager`负责事务的开启和提交操作。当你在一个方法上使用`@Transactional`注解时，Spring会通过`TransactionManager`自动管理事务的开启和提交。
- **事务的回滚：** 如果在事务处理过程中发生异常或出现错误，`TransactionManager`会将事务标记为回滚状态，确保事务不会提交，并将所有的更改都回滚到事务开始之前的状态。
- **多数据源管理：** 对于多数据源的情况，不同数据源可以有不同的`TransactionManager`，以确保每个数据源的事务行为得以正确管理。
`PlatformTransactionManager`是`TransactionManager`的一个常见实现，用于管理不同类型的事务（例如，JDBC、JPA、Hibernate等）。常见的`PlatformTransactionManager`实现包括`DataSourceTransactionManager`（用于JDBC事务）和`JpaTransactionManager`（用于JPA事务）。
2. **TransactionTemplate：**
`TransactionTemplate`是Spring框架中的一个辅助类，它简化了在事务环境中执行操作的编程。`TransactionTemplate`封装了事务的开启、提交、回滚等操作，使你不必手动处理这些事务操作，从而减少了重复的模板代码。它的一些主要功能包括：
- **自动事务处理：** 你可以将需要在事务环境中执行的操作传递给`TransactionTemplate`，它会自动为你管理事务的开启、提交和回滚。
- **异常处理：** `TransactionTemplate`可以捕获操作中的异常，自动将事务标记为回滚状态，以确保在出现异常时事务不会提交，并会回滚到事务开始前的状态。
- **线程绑定：** `TransactionTemplate`会自动处理线程和事务之间的绑定，确保在操作方法中不需要考虑事务的开启和关闭。
通过使用`TransactionTemplate`，你可以更专注于业务逻辑的实现，而无需过多关注事务的细节。
综上所述，`TransactionManager`负责底层事务管理，而`TransactionTemplate`是在上层封装，为事务操作提供了更便捷的方式。你可以根据项目的需要，选择使用`TransactionManager`、`TransactionTemplate`或两者结合使用，以实现有效的事务管理。
