批处理优化:从稳定性、性能、数据一致性、健壮性、可观测性五大维度,优化批量操作
摘要:本文围绕 "批量操作" 展开,从稳定性(分批处理、重试机制、中断恢复)、性能(批量插入、SQL 优化、并发编程、异步任务)、数据一致性(事务管理、分布式锁 / 乐观锁)、健壮性(提前参数校验、细粒度异常处理)、可观测性(日志记录、监控、返回结果优化)五大维度,系统提供批处理操作的优化方案。
1. 题目批量管理
需求分析
为了提高管理效率,需提供批量操作功能,例如:【管理员】批量向题库添加题目
基础方案设计
前端设计
前端需要允许用户多选内容,并且向后端传递要批量操作的数据 id 列表。
批量向题库添加题目:选中多条题目后,点击操作按钮,弹窗让用户选择要添加的题库。
后端设计
后端通过 循环 依次调用数据库完成操作。
注意:由于是批量操作,需要使用事务,有任何失败都会抛出异常并回滚。
基础后端开发
批量向题库添加题目
添加前需要校验题目和题库是否存在,只添加合法的题目:
@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, "向题库添加题目失败");}}
}
2. 批处理操作优化 ⭐⭐⭐
稳定性
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()
方法获取到了当前实现类的代理对象,来调用事务方法。
为什么要这么做呢? 因为 Spring 事务依赖于代理机制,而内部调用通过 this
直接调用方法,不会通过 Spring 的代理,因此不会触发事务。
注意,使用 AopContext.currentProxy()
方法时必须要在启动类添加下面的注解开启切面自动代理:
@EnableAspectJAutoProxy(proxyTargetClass = true, exposeProxy = true)
2. 重试机制
对于可能由于网络不稳定等临时原因偶发失败的操作,可以设计 重试机制 提高系统的稳定性,适用于执行时间很长的任务。
注意,重试的过程中要记录日志,并且重试次数要有一个上限 。示例代码如下:
int retryCount = 3;
for (int i = 0; i < retryCount; i++) {try {// 执行插入操作// 成功则跳出重试循环break; } catch (Exception e) {log.warn("插入失败,重试次数: {}", i + 1);if (i == retryCount - 1) {throw new BusinessException(ErrorCode.OPERATION_ERROR, "多次重试后仍然失败");}}
}
3. 中断恢复
如果在批量插入过程中由于某种原因(如数据库宕机、服务器重启)导致批处理中断,建议设计一种机制来进行 增量恢复。比如可以为每次操作打上批次标记,在操作未完成时记录操作状态(如部分题目成功添加),并在恢复时继续执行未完成的操作。
可以设计一个数据库表存储批次的状态:
create table question_batch_status (batch_id bigint primary key,question_bank_id bigint,total_questions int,processed_questions int,status varchar(20) -- running, completed, failed
);
通过该表可以跟踪每次批处理的进度,并在失败时根据批次继续处理。
性能优化
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 优化原则:不要使用 select *
来查询数据,只查出需要的字段即可
比如:
// 检查题目 id 是否存在
LambdaQueryWrapper<Question> questionLambdaQueryWrapper = Wrappers.lambdaQuery(Question.class).select(Question::getId).in(Question::getId, questionIdList);
List<Question> questionList = questionService.list(questionLambdaQueryWrapper);
由于返回的值只有 id 一列,可直接转为 Long 列表,无需让框架封装结果为对象,减少内存占用:
// 合法的题目 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 线程池来并发执行任务。该线程池特别适合需要分治法来处理的大量并发任务,支持递归任务拆分。Java 8 中的并行流默认也是使用了 ForkJoinPool 进行并发处理
ForkJoinPool 的主要特性:
工作窃取算法(Work-Stealing):线程可以从其他线程的工作队列中“窃取”任务,以提高 CPU 的使用率和程序的并行性。
递归任务处理:支持将大任务拆分为多个小任务并行执行,然后再将结果合并。
💡 但是要注意,CompletableFuture 默认使用的是 ForkJoinPool.commonPool()
方法得到的线程池,这是一个全局共享的线程池,如果有多种不同的任务都依赖该线程池进行处理,可能会导致资源争抢、代码阻塞等不确定的问题。所以建议针对每种任务,自定义线程池来处理,实现线程池资源的隔离。
Java 内置了很多种不同的线程池,比如单线程的线程池、固定线程的线程池、自定义线程池等等,一般情况下我们会根据业务和资源情况 自定义线程池。
// 自定义线程池
ThreadPoolExecutor customExecutor = new ThreadPoolExecutor(4, // 核心线程数10, // 最大线程数60L, // 线程空闲存活时间TimeUnit.SECONDS, // 存活时间单位new LinkedBlockingQueue<>(1000), // 阻塞队列容量new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略:由调用线程处理任务
);
自定义线程池里有那么多参数,应该如何设置呢?
此处大家只要记住一个公式:
对于计算密集型任务(消耗 CPU 资源), 设置核心线程数为
n+1
或者n
(n 为 CPU 核心数),可以充分利用 CPU, 多一个线程是为了可以在某些线程短暂阻塞或执行调度时,确保有足够的线程保持 CPU 繁忙,最大化 CPU 的利用率。对于 IO 密集型任务(消耗 IO 资源),可以增大核心线程数为 CPU 核心数的 2 - 4 倍,可以提升并发执行任务的数量。
对于批量添加题目功能,和数据库交互频繁,属于 IO 密集型任务,可以给自定义线程池更大的核心线程数。引入并发编程后的代码:
// 自定义线程池
ThreadPoolExecutor customExecutor = new ThreadPoolExecutor(20, // 核心线程数50, // 最大线程数60L, // 线程空闲存活时间TimeUnit.SECONDS, // 存活时间单位new LinkedBlockingQueue<>(10000), // 阻塞队列容量new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略:由调用线程处理任务
);
// 用于保存所有批次的 CompletableFuture
List<CompletableFuture<Void>> futures = new ArrayList<>();
// 分批处理避免长事务,假设每次处理 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();// 异步处理每批数据并添加到 futures 列表CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {questionBankQuestionService.batchAddQuestionsToBankInner(questionBankQuestions);}, customExecutor);futures.add(future);
}
// 等待所有批次操作完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
// 关闭线程池
customExecutor.shutdown();
注意,虽然并发编程能够提升性能,但也会占用更多的资源,并且给系统引入更多的不确定性。比如某个任务出现异常时,其他任务可能正在执行,产生不确定的影响。对此,可以根据情况给异步任务补充异常处理行为,通过 exceptionally
方法就能实现,示例代码如下:
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {questionBankQuestionService.batchAddQuestionsToBankInner(questionBankQuestions);
}, customExecutor).exceptionally(ex -> {log.error("批处理任务执行失败", ex);return null;
});
4. 异步任务优化⭐
将批量操作的处理变成提交一个后台任务,提交后台任务后,接口可以直接给前端返回已提交的任务 id。后台可以根据情况选择时机去执行之前提交的后台任务(如通过定时任务或者消息队列)。
业务流程对应的示例代码如下:
@PostMapping("/start")
public String startTask() {// 生成唯一任务 idString taskId = UUID.randomUUID().toString();// 异步提交或执行任务executeLongRunningTask(taskId);// 返回任务 idreturn taskId;
}
前端可以通过轮询调用接口、WebSocket、SSE 等方式得知任务的执行进度。
实现异步任务
1)立即执行异步任务
可以直接给要异步执行的方法加上 Spring 提供的 @Async
注解,或者手动使用 CompletableFuture
来异步执行任务。
@Async
@Override
@Transactional(rollbackFor = Exception.class)
public void batchAddQuestionsToBankInner(List<QuestionBankQuestion> questionBankQuestions) {// ...
}
2)定时执行
可将任务信息存数据库中,通过 Spring Scheduler 定时任务持续扫描数据库 未执行的任务 执行。
@Component
public class TaskScheduler {// 每30秒执行一次(cron表达式:秒 分 时 日 月 周)@Scheduled(cron = "0/30 * * * * ?")public void scanAndExecuteTasks() {System.out.println("开始扫描待执行任务...");// 处理任务 }
}
3)通过MQ进行任务分发
对于长时间的批量任务,还可考虑使用 消息队列来异步处理任务。将任务放入消息队列,由消费者(后台服务)异步执行任务。
示例任务提交代码:
@PostMapping("/start")
public String startTask() {// 生成唯一任务 idString taskId = UUID.randomUUID().toString();// 将任务发送到消息队列messageQueueService.sendMessage(new TaskMessage(taskId));
// 返回任务 idreturn taskId;
}
后台消费者接收到消息后处理任务:
@RabbitListener(queues = "batch-task-queue")
public void processBatchTask(TaskMessage taskMessage) {Long taskId = taskMessage.getTaskId();// 查询数据库,根据 taskId 获取任务信息Task task = getById(taskId);// 执行任务doSomething(task);// 执行完成后,记得更新任务的状态
}
数据一致性
1. 事务管理⭐
我们目前已经使用了 @Transactional(rollbackFor = Exception.class)
来保证数据一致性。任意一步操作失败,整个事务会回滚,确保数据一致性。
2. 并发管理⭐
在高并发场景下,如果多个管理员同时向同一个题库添加题目,可能会导致冲突或性能问题。为了解决并发问题,确保数据一致性和稳定性,可以有 2 种常见的策略:
1)增加 分布式锁 来防止同一个接口在同一时间被多人同时操作,如使用 Redis + Redisson 实现分布式锁。
2)如果要精细地对某个数据进行并发控制,可以选用 乐观锁。比如通过给 QuestionBank
表增加一个 version
字段,在更新时检查版本号是否一致,确保对同一个题库的并发操作不会相互干扰。
// 更新题库前,先查询版本号
QuestionBank questionBank = questionBankService.getById(questionBankId);
Long currentVersion = questionBank.getVersion();
// 更新时,检查版本号是否一致
int rowsAffected = questionBankService.updateVersionById(questionBankId, currentVersion);
if (rowsAffected == 0) {throw new BusinessException(ErrorCode.CONCURRENT_MODIFICATION, "数据已被其他用户修改");
}
💡 在 MySQL 中,还可以采用
SELECT ... FOR UPDATE
来强行锁定某一行数据,直到当前事务提交或回滚之前,防止其他事务对这行数据进行修改。本项目中,由于关联表有唯一键约束,保证不会重复,所以不需要用这种方案。
健壮性
健壮性是指系统在面对 异常情况或不合法输入 时仍能表现出合理的行为。一个健壮的系统能够 预见和处理异常,并且即使发生错误,也不会崩溃或产生不可预期的行为。
1. 参数校验提前
可以在调用数据库之前就对参数进行校验,这样可以减少不必要的数据库操作开销,不用等到数据库操作时再抛出异常。
在现有的添加题目到题库的代码中,我们已经提前对参数进行了非空校验,并且会提前检查题目和题库是否存在,这是很好的。但是我们还没有校验哪些题目已经添加到题库中,对于这些题目,不必再执行插入关联记录的数据库操作。
需要补充的代码如下:
// ...// 检查哪些题目还不存在于题库中,避免重复插入
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, "向题库添加题目失败");
}
可观测性
可观测性的关键在于以下三个方面:
可见性:系统需要能够报告它的内部状态。这个优化方案通过返回
BatchAddResult
提供了丰富的状态反馈。追踪性:通过详细的错误原因和具体失败项,可以轻松地追踪问题源头。
诊断性:明确的反馈信息有助于快速诊断问题,而不仅是提供一个简单的 "成功" 或 "失败"。
1. 日志记录⭐
高并发场景,批量操作可能会出现意外问题,建议记录日志:包括成功、失败的题目,便于排查。
//比如记录日志
log.error("数据库唯一键冲突或违反其他完整性约束, 错误信息: {}", e.getMessage());
2. 监控
监控是实现可观测性的主流手段,可对服务器、请求、以及项目中引入的各种组件进行监控。
常用的监控工具有 Grafana,如果你给项目引入了某个技术组件,一般都会自带监控,比如项目调用数据库的情况可以通过 Druid 监控、Elasticsearch 可以通过 Kibana 监控等等、Spring Boot 内置了 Spring Boot Actuator 来监控应用运行状态等。
3. 返回值优化
目前我们的方法返回的是 void
,这意味着在执行过程中没有明确反馈操作的结果。为了提升可观测性,我们可以根据任务的执行状态返回更加详细的结果,帮助调用者了解任务的执行情况。
可以定义一个返回结果对象,包含每个题目的处理状态、成功和失败的数量,以及失败的原因。
public class BatchAddResult {private int total;private int successCount;private int failureCount;private List<String> failureReasons;
}
在批量操作中出现问题时,可以不抛出异常并中断其他的操作,只是记录部分失败的操作情况。这样一来,管理员可以知道哪些题目操作成功、哪些失败,更好地进行后续处理。
public BatchAddResult batchAddQuestionsToBank(List<Long> questionIdList, Long questionBankId, User loginUser) {BatchAddResult result = new BatchAddResult();result.setTotal(questionIdList.size());
// 执行批量插入逻辑for (Long questionId : questionIdList) {try {// 插入操作saveQuestionToBank(questionId, questionBankId, loginUser);result.setSuccessCount(result.getSuccessCount() + 1);} catch (Exception e) {result.setFailureCount(result.getFailureCount() + 1);result.getFailureReasons().add("题目ID " + questionId + " 插入失败:" + e.getMessage());}}
return result; // 返回批量处理的结果
}
学到如此多优化方法,大家可以自己根据情况选用。