如何用Java 11 Stream分块调用@Async标注的saveBulk批量保存？

当然可以用Java 11 Stream实现这个需求，还能结合异步特性做到高效处理。核心思路是把大列表拆分成固定大小的子批次，通过Stream遍历每个子批次调用异步的saveBulk方法，最后统一等待所有异步任务完成。

实现步骤与代码示例 #

先通过IntStream生成批次索引，再根据索引截取原列表的子批次，同时收集异步任务的返回结果：

List<User> userList = userService.selectAll(); // 百万条数据
int batchSize = 1000; // 根据数据库性能、连接池配置调整批次大小

// 生成所有批次并调用异步保存方法
List<CompletableFuture<Void>> futures = IntStream.range(0, (userList.size() + batchSize - 1) / batchSize)
        .mapToObj(i -> {
            int start = i * batchSize;
            int end = Math.min(start + batchSize, userList.size());
            // 截取子批次，若需要独立列表可改用new ArrayList<>(userList.subList(start, end))
            List<User> batch = userList.subList(start, end);
            return customerService.saveBulk(batch);
        })
        .collect(Collectors.toList());

// 等待所有异步任务完成，处理异常
try {
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
} catch (InterruptedException | ExecutionException e) {
    Thread.currentThread().interrupt();
    throw new RuntimeException("批量保存失败", e);
}

关键细节说明 #

• 批次大小选择：建议选1000-5000条，避免单次插入数据过多导致数据库压力过载，具体数值可根据实际测试调整。
• 异步任务管理：用CompletableFuture.allOf()统一等待所有任务完成，比逐个调用get()更高效，不会导致任务串行执行。
• 异常处理：必须捕获InterruptedException（线程被中断）和ExecutionException（异步任务内部抛异常），根据业务需求做日志记录、重试或告警。
• 子列表注意事项：List.subList()返回的是原列表的视图，若原列表后续被修改会影响子批次，若需要独立列表可创建新的ArrayList包装子列表。

可复用的分批工具封装（可选） #

如果需要在多处复用分批逻辑，可以封装一个工具方法：

public static <T> Stream<List<T>> batchStream(List<T> list, int batchSize) {
    if (list == null || list.isEmpty()) {
        return Stream.empty();
    }
    int totalBatches = (list.size() + batchSize - 1) / batchSize;
    return IntStream.range(0, totalBatches)
            .mapToObj(i -> {
                int start = i * batchSize;
                int end = Math.min(start + batchSize, list.size());
                return new ArrayList<>(list.subList(start, end));
            });
}

使用时会更简洁：

List<CompletableFuture<Void>> futures = batchStream(userList, 1000)
        .map(customerService::saveBulk)
        .collect(Collectors.toList());

CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

内容的提问来源于stack exchange，提问作者jlc488