java使用CompletableFuture分批处理任务实现
作者：呦，又写BUG呢
本文主要介绍了java使用CompletableFuture分批处理任务实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
一、无返回值任务函数
// 数据分批
List<List<StatisticsDTO>> batches = Lists.partition(statisticsList, BATCH_SIZE);
List<CompletableFuture<Void>> futures = new ArrayList<>(batches.size());
// 数据处理
for (int i = 0; i < batches.size(); i++) {
logger.info("批次 " + i + " 开始处理...");
String logId = LogIdThreadLocal.getLogId(); // 传递主线程的 logId
List<StatisticsDTO> batchData = batches.get(i);
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
try {
LogIdThreadLocal.setLogId(logId);
processBatch(batchData);
} finally {
LogIdThreadLocal.clean();
}
});
futures.add(future);
}
// 等待所有的异步任务完成
CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
allOf.join();
二、带返回值任务函数
// 数据分批
List<List<StatisticsDTO>> batches = Lists.partition(statisticsList, BATCH_SIZE);
List<CompletableFuture<List<StatisticsDTO>>> futures = new ArrayList<>(batches.size());
// 数据处理
for (int i = 0; i < batches.size(); i++) {
logger.info("批次 " + i + " 开始处理...");
String logId = LogIdThreadLocal.getLogId(); // 传递主线程的 logId
List<StatisticsDTO> batchData = batches.get(i);
CompletableFuture<List<DoctorAvatarAnalysisDTO>> future = CompletableFuture.supplyAsync(() -> {
try {
LogIdThreadLocal.setLogId(logId);
return processBatch(batchData);
} finally {
LogIdThreadLocal.clean();
}
});
futures.add(future);
}
// 等待所有 CF 完成并合并结果
CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
List<StatisticsDTO> result = allOf.thenApply(
v -> futures.stream().map(CompletableFuture::join).flatMap(List::stream).collect(Collectors.toList())
).join();
到此这篇关于java使用CompletableFuture分批处理任务实现的文章就介绍到这了,更多相关java CompletableFuture分批处理内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！