当前位置：首页 > news >正文

Spring Boot + MyBatis-Plus 单数据源多线程事务一致性实践

news 2025/9/11 5:14:51

Spring Boot + MyBatis-Plus 单数据源多线程事务一致性实践

关键词：Spring Boot、MyBatis-Plus、单数据源、多线程、事务一致性、编程式事务、CompletableFuture

一、问题背景

在 Spring Boot + MyBatis-Plus 的单数据源应用中，如果业务需要并行处理大量数据，往往会在 Service 层把任务拆分成若干子任务，再丢进线程池并行执行。然而，原生的 @Transactional 是基于 ThreadLocal 的，只能管理当前线程的事务；子线程里的操作对主线程事务是不可见的，这就导致：

子线程异常不会触发主线程回滚；
子线程成功提交后，主线程再回滚，数据出现“一半成功一半失败”。

因此，需要一种跨线程且最终一致性的解决方案。

二、核心思路

放弃声明式事务（@Transactional），改用编程式事务。
每个子线程独立开启一个新事务，拿到自己的 Connection 和 TransactionStatus。
所有子线程执行完毕后，由主线程统一判断结果：
- 全部成功 → 统一提交；
- 任一失败 → 统一回滚。

三、实现步骤

3.1 依赖坐标（已包含 Spring Boot、MyBatis-Plus）

<dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.5.5</version>
</dependency>

3.2 线程安全的数据结构

名称	说明
`List<TransactionStatus>`	存放每个子线程的事务句柄，需线程安全。
`AtomicBoolean`	标记是否有子线程抛异常。
`ExecutorService`	独立线程池，避免与业务线程混杂。

3.3 核心工具类

package com.example.tx;import lombok.RequiredArgsConstructor;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;import javax.sql.DataSource;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;@Component
@RequiredArgsConstructor
public class MultiThreadTxTemplate {private final DataSource dataSource;public void runInNewTx(List<Runnable> tasks, Executor executor) throws Exception {DataSourceTransactionManager txManager = new DataSourceTransactionManager(dataSource);List<TransactionStatus> txStatusList = Collections.synchronizedList(new ArrayList<>());AtomicBoolean hasError = new AtomicBoolean(false);List<CompletableFuture<Void>> futures = tasks.stream().map(task -> CompletableFuture.runAsync(() -> {// 1. 新建事务DefaultTransactionDefinition def = new DefaultTransactionDefinition();TransactionStatus status = txManager.getTransaction(def);txStatusList.add(status);try {task.run();} catch (Throwable ex) {hasError.set(true);throw new RuntimeException(ex);}}, executor)).toList();// 2. 等待全部任务结束try {CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();} catch (CompletionException ex) {hasError.set(true);}// 3. 统一提交或回滚if (hasError.get()) {txStatusList.forEach(txManager::rollback);throw new RuntimeException("批量任务执行失败，已统一回滚");} else {txStatusList.forEach(txManager::commit);}}
}

3.4 使用示例

@Service
@RequiredArgsConstructor
public class UserBatchService {private final UserMapper userMapper;private final MultiThreadTxTemplate txTemplate;private final ExecutorService executor = Executors.newFixedThreadPool(4);@SneakyThrowspublic void batchInsert(List<User> users) {List<Runnable> tasks = users.stream().<Runnable>map(u -> () -> userMapper.insert(u)).toList();txTemplate.runInNewTx(tasks, executor);}
}

3.5 单元测试

@SpringBootTest
class UserBatchServiceTest {@AutowiredUserBatchService userBatchService;@AutowiredUserMapper userMapper;@Testvoid shouldRollbackWhenAnySubTaskFail() {List<User> users = List.of(new User(1, "A"),new User(2, "B"),new User(3, "C"),new User(4, "D"));// 模拟第三个任务失败Mockito.doThrow(new RuntimeException("mock error")).when(userMapper).insert(users.get(2));assertThrows(RuntimeException.class, () -> userBatchService.batchInsert(users));// 断言：数据库应无任何记录assertEquals(0, userMapper.selectCount(null));}
}

四、常见问题与对策

问题	现象	解决
连接池耗尽	子线程过多	合理设置线程池大小，或分批执行
主线程事务污染	主线程也开启了事务	主线程不要加 `@Transactional`，或手动保存/恢复上下文
线程池复用导致旧事务残留	第二次调用时未清理	每次调用都 new 一个 `DataSourceTransactionManager`，无残留