批处理操作优化思路

基础功能——批量向题库添加题目

@Override
@Transactional(rollbackFor = Exception.class)
public void batchAddQuestionsToBank(List<Long> questionIdList, Long questionBankId, User loginUser) {// 参数校验ThrowUtils.throwIf(CollUtil.isEmpty(questionIdList), ErrorCode.PARAMS_ERROR, "题目列表为空");ThrowUtils.throwIf(questionBankId == null || questionBankId <= 0, ErrorCode.PARAMS_ERROR, "题库非法");ThrowUtils.throwIf(loginUser == null, ErrorCode.NOT_LOGIN_ERROR);// 检查题目 id 是否存在List<Question> questionList = questionService.listByIds(questionIdList);// 合法的题目 idList<Long> validQuestionIdList = questionList.stream().map(Question::getId).collect(Collectors.toList());ThrowUtils.throwIf(CollUtil.isEmpty(validQuestionIdList), ErrorCode.PARAMS_ERROR, "合法的题目列表为空");// 检查题库 id 是否存在QuestionBank questionBank = questionBankService.getById(questionBankId);ThrowUtils.throwIf(questionBank == null, ErrorCode.NOT_FOUND_ERROR, "题库不存在");// 执行插入for (Long questionId : validQuestionIdList) {QuestionBankQuestion questionBankQuestion = new QuestionBankQuestion();questionBankQuestion.setQuestionBankId(questionBankId);questionBankQuestion.setQuestionId(questionId);questionBankQuestion.setUserId(loginUser.getId());boolean result = this.save(questionBankQuestion);if (!result) {throw new BusinessException(ErrorCode.OPERATION_ERROR, "向题库添加题目失败");}}
}

健壮性优化

健壮性是指系统在面对 异常情况或不合法输入 时仍能表现出合理的行为。一个健壮的系统能够 预见和处理异常，并且即使发生错误，也不会崩溃或产生不可预期的行为。

1、参数校验提前

在现有的添加题目到题库的代码中，我们已经提前对参数进行了非空校验，并且会提前检查题目和题库是否存在，这是很好的。但是我们还没有校验哪些题目已经添加到题库中，对于这些题目，不必再执行插入关联记录的数据库操作。

// 检查题库 id 是否存在
// ...// 检查哪些题目还不存在于题库中，避免重复插入
LambdaQueryWrapper<QuestionBankQuestion> lambdaQueryWrapper = Wrappers.lambdaQuery(QuestionBankQuestion.class).eq(QuestionBankQuestion::getQuestionBankId, questionBankId).in(QuestionBankQuestion::getQuestionId, validQuestionIdList);
List<QuestionBankQuestion> existQuestionList = this.list(lambdaQueryWrapper);
// 已存在于题库中的题目 id
Set<Long> existQuestionIdSet = existQuestionList.stream().map(QuestionBankQuestion::getQuestionId).collect(Collectors.toSet());
// 已存在于题库中的题目 id，不需要再次添加
validQuestionIdList = validQuestionIdList.stream().filter(questionId -> {return !existQuestionIdSet.contains(questionId);
}).collect(Collectors.toList());
ThrowUtils.throwIf(CollUtil.isEmpty(validQuestionIdList), ErrorCode.PARAMS_ERROR, "所有题目都已存在于题库中");// 执行插入
// ...

2、异常处理

目前虽然已经对每一次插入操作的结果都进行了判断，并且抛出自定义异常，但是有些特殊的异常并没有被捕获。

可以进一步细化异常处理策略，考虑更细粒度的异常分类，不同的异常类型可以通过不同的方式处理，例如：

数据唯一键重复插入问题，会抛出 DataIntegrityViolationException。
数据库连接问题、事务问题等导致操作失败时抛出 DataAccessException。
其他的异常可以通过日志记录详细错误信息，便于后期追踪（全局异常处理器也有这个能力）。

try {boolean result = this.save(questionBankQuestion);if (!result) {throw new BusinessException(ErrorCode.OPERATION_ERROR, "向题库添加题目失败");}
} catch (DataIntegrityViolationException e) {log.error("数据库唯一键冲突或违反其他完整性约束，题目 id: {}, 题库 id: {}, 错误信息: {}",questionId, questionBankId, e.getMessage());throw new BusinessException(ErrorCode.OPERATION_ERROR, "题目已存在于该题库，无法重复添加");
} catch (DataAccessException e) {log.error("数据库连接问题、事务问题等导致操作失败，题目 id: {}, 题库 id: {}, 错误信息: {}",questionId, questionBankId, e.getMessage());throw new BusinessException(ErrorCode.OPERATION_ERROR, "数据库操作失败");
} catch (Exception e) {// 捕获其他异常，做通用处理log.error("添加题目到题库时发生未知错误，题目 id: {}, 题库 id: {}, 错误信息: {}",questionId, questionBankId, e.getMessage());throw new BusinessException(ErrorCode.OPERATION_ERROR, "向题库添加题目失败");
}

稳定性优化

1、避免长事务问题

批量操作中，一次性处理过多数据会导致事务过长，影响数据库性能。可以通过分批处理来避免长事务问题，确保部分数据异常不会影响整个批次的数据保存。

假设操作 10w 条数据，其中有 1 条数据操作异常，如果是长事务，那么修改的 10w 条数据都需要回滚，而分批事务仅需回滚一批既可，降低长事务带来的资源消耗，同时也提升了稳定性。

@Override
@Transactional(rollbackFor = Exception.class)
public void batchAddQuestionsToBankInner(List<QuestionBankQuestion> questionBankQuestions) {for (QuestionBankQuestion questionBankQuestion : questionBankQuestions) {long questionId = questionBankQuestion.getQuestionId();long questionBankId = questionBankQuestion.getQuestionBankId();try {boolean result = this.save(questionBankQuestion);ThrowUtils.throwIf(!result, ErrorCode.OPERATION_ERROR, "向题库添加题目失败");} catch (DataIntegrityViolationException e) {log.error("数据库唯一键冲突或违反其他完整性约束，题目 id: {}, 题库 id: {}, 错误信息: {}",questionId, questionBankId, e.getMessage());throw new BusinessException(ErrorCode.OPERATION_ERROR, "题目已存在于该题库，无法重复添加");} catch (DataAccessException e) {log.error("数据库连接问题、事务问题等导致操作失败，题目 id: {}, 题库 id: {}, 错误信息: {}",questionId, questionBankId, e.getMessage());throw new BusinessException(ErrorCode.OPERATION_ERROR, "数据库操作失败");} catch (Exception e) {// 捕获其他异常，做通用处理log.error("添加题目到题库时发生未知错误，题目 id: {}, 题库 id: {}, 错误信息: {}",questionId, questionBankId, e.getMessage());throw new BusinessException(ErrorCode.OPERATION_ERROR, "向题库添加题目失败");}}
}

在原方法中批量生成题目，并且调用上述事务方法：

// 分批处理避免长事务，假设每次处理 1000 条数据
int batchSize = 1000;
int totalQuestionListSize = validQuestionIdList.size();
for (int i = 0; i < totalQuestionListSize; i += batchSize) {// 生成每批次的数据List<Long> subList = validQuestionIdList.subList(i, Math.min(i + batchSize, totalQuestionListSize));List<QuestionBankQuestion> questionBankQuestions = subList.stream().map(questionId -> {QuestionBankQuestion questionBankQuestion = new QuestionBankQuestion();questionBankQuestion.setQuestionBankId(questionBankId);questionBankQuestion.setQuestionId(questionId);questionBankQuestion.setUserId(loginUser.getId());return questionBankQuestion;}).collect(Collectors.toList());// 使用事务处理每批数据QuestionBankQuestionService questionBankQuestionService = (QuestionBankQuestionServiceImpl) AopContext.currentProxy();questionBankQuestionService.batchAddQuestionsToBankInner(questionBankQuestions);
}

注意：要通过 AopContext.currentProxy() 方法获取到了当前实现类的代理对象，来调用事务方法。这是因为在调用的是同一个类中带有 @Transactional 注解的方法，如果不通过 AopContext.currentProxy() 获取代理对象而是直接 this.xxx() 调用的话，事务不会生效！

具体来说，Spring 的 @Transactional 是通过 AOP（面向切面编程）代理实现的。简单点说，Spring 会创建一个你的类的“代理类”，在调用你方法的时候，会先执行一些“切面逻辑”（比如事务开启），然后再调你真正的方法。

batchAddQuestionsToBankInner() 是有事务的，但是你是在同一个类内部直接调用它，这样的话 Spring 就绕过了 AOP 代理机制，等于你是直接调用原方法，事务根本不会被管理！

只有通过 AopContext.currentProxy() 让batchAddQuestionsToBank调用 代理对象 ，这才会触发 @Transactional 逻辑。

原类:        QuestionBankServiceImpl
代理类:      Proxy(QuestionBankServiceImpl)注入的是这个 Proxy，而不是原来的类！

调用流程应该长这样：

外部调用 --> Proxy（事务处理） --> 原方法

但是this.batchAddQuestionsToBankInner() 的调用流程是：

this（原类） --> batchAddQuestionsToBankInner（原类方法，绕过代理）

性能优化

1、批量操作

当前代码中，每个题目是单独插入数据库的，这会产生频繁的数据库交互。

大多数 ORM 框架和数据库驱动都支持批量插入，可以通过批量插入来优化性能，比如 MyBatis Plus 提供了 saveBatch 方法。

@Override
@Transactional(rollbackFor = Exception.class)
public void batchAddQuestionsToBankInner(List<QuestionBankQuestion> questionBankQuestions) {try {boolean result = this.saveBatch(questionBankQuestions);ThrowUtils.throwIf(!result, ErrorCode.OPERATION_ERROR, "向题库添加题目失败");} catch (DataIntegrityViolationException e) {log.error("数据库唯一键冲突或违反其他完整性约束, 错误信息: {}", e.getMessage());throw new BusinessException(ErrorCode.OPERATION_ERROR, "题目已存在于该题库，无法重复添加");} catch (DataAccessException e) {log.error("数据库连接问题、事务问题等导致操作失败, 错误信息: {}", e.getMessage());throw new BusinessException(ErrorCode.OPERATION_ERROR, "数据库操作失败");} catch (Exception e) {// 捕获其他异常，做通用处理log.error("添加题目到题库时发生未知错误，错误信息: {}", e.getMessage());throw new BusinessException(ErrorCode.OPERATION_ERROR, "向题库添加题目失败");}
}

批量操作的好处：

降低了数据库连接和提交的频率。
避免频繁的数据库交互，减少 I/O 操作，显著提高性能。

2、SQL 优化

在操作数据库时，可以使用一些 SQL 优化的技巧。

其中，有一个最基本的 SQL 优化原则，不要使用 select * 来查询数据，只查出需要的字段即可。

// 检查题目 id 是否存在
LambdaQueryWrapper<Question> questionLambdaQueryWrapper = Wrappers.lambdaQuery(Question.class).select(Question::getId).in(Question::getId, questionIdList);
List<Question> questionList = questionService.list(questionLambdaQueryWrapper);

由于返回的值只有 id 一列，还可以直接转为 Long 列表，不需要让框架封装结果为 Question 对象了，减少内存占用：

// 合法的题目 id
List<Long> validQuestionIdList = questionService.listObjs(questionLambdaQueryWrapper, obj -> (Long) obj);
ThrowUtils.throwIf(CollUtil.isEmpty(validQuestionIdList), ErrorCode.PARAMS_ERROR, "合法的题目列表为空");

3、并发编程

已经将操作分批处理，在操作较多、追求处理时间的情况下，可以通过并发编程让每批操作同时执行，而不是一批处理完再执行下一批，能够大幅提升性能。

Java 中，可以利用并发包中的 CompletableFuture + 线程池来并发处理多个任务。

CompletableFuture 是 Java 8 中引入的一个类，用于表示异步操作的结果。它是 Future 的增强版本，不仅可以表示一个异步计算，还可以对异步计算的结果进行组合、转换和处理，实现异步任务的编排。

比如下列代码，将任务拆分为多个子任务，并发执行，最后通过 CompletableFuture.allOf 方法阻塞等待，只有所有的子任务都完成，才会执行后续代码：

List<CompletableFuture<Void>> futures = new ArrayList<>();for (List<Long> subList : splitList(validQuestionIdList, 1000)) {CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {processBatch(subList, questionBankId, loginUser);});futures.add(future);
}// 等待所有任务完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

CompletableFuture 默认使用 Java 7 引入的 ForkJoinPool 线程池来并发执行任务。该线程池特别适合需要分治法来处理的大量并发任务，支持递归任务拆分。

ForkJoinPool 的主要特性：

工作窃取算法（Work-Stealing）：线程可以从其他线程的工作队列中“窃取”任务，以提高 CPU 的使用率和程序的并行性。
递归任务处理：支持将大任务拆分为多个小任务并行执行，然后再将结果合并。

自定义线程池

// 自定义线程池
ThreadPoolExecutor customExecutor = new ThreadPoolExecutor(4,                         // 核心线程数10,                        // 最大线程数60L,                       // 线程空闲存活时间TimeUnit.SECONDS,           // 存活时间单位new LinkedBlockingQueue<>(1000),  // 阻塞队列容量new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：由调用线程处理任务
);

这样做，相当于是开了一个快递打包工厂：

有 10 万个包裹（题目）
每 1000 个包裹组成一个大箱子（批次）
工厂最多一次同时开 50 个打包员（最大线程数）
每个打包员一次只处理一个大箱子（一个批次）
每个线程负责处理一个完整的批次。

虽然并发编程能够提升性能，但也会占用更多的资源，并且给系统引入更多的不确定性。比如某个任务出现异常时，其他任务可能正在执行，产生不确定的影响。对此，可以根据情况给异步任务补充异常处理行为，通过 exceptionally 方法就能实现，示例代码如下：

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {questionBankQuestionService.batchAddQuestionsToBankInner(questionBankQuestions);
}, customExecutor).exceptionally(ex -> {log.error("批处理任务执行失败", ex);return null;
});