首次使用CompletableFuture结合Lambda API批量并行处理的正确性验证

需求背景 #

首次使用CompletableFuture，需将接收的一组输入按每5个为一批，通过Lambda API并行处理后展示，确认当前代码是否正确实现批量并行调用逻辑。

用户提供的代码 #

List<Info> listOfInputs = new LinkedList<>();

List<CompletableFuture<List<Info>>> futures = new ArrayList<>();
try {
    for (int iterator = 0; iterator < inputs.length; iterator = iterator + 5)) {
        int finalIterator = iterator;
        CompletableFuture<List<Info>> future = CompletableFuture.supplyAsync(() -> {
            return getAllInputs(Arrays.copyOfRange(inputs, finalIterator,
                                    Math.min(inputs.length, finalIterator + 5)));
        });
        futures.add(future);
    }
    CompletableFuture<Void> allFutures = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0]));
    allFutures.join();

    for (CompletableFuture<List<Info>> future : futures) {
        List<Info> info = future.get();
        for (int i = 0; i < info.size(); i++) {
            listOfInputs.add(info.get(i));
        }
    }

    listOfInputs.stream().map(infoEntry -> {
            sout("Hello" + infoEntry) });  
}                     

核心逻辑正确性判断 #

你的代码核心的批量并行调用逻辑是正确的，具体体现在：

• 批量拆分：通过步长为5的循环，结合Arrays.copyOfRange和Math.min，正确实现了每5个输入为一批的拆分，同时处理了最后一批不足5个的边界情况。
• 并行执行：使用CompletableFuture.supplyAsync将每批任务提交到默认线程池并行执行，再通过CompletableFuture.allOf等待所有任务完成，确保所有批量处理都执行完毕后再收集结果。

代码存在的问题 #

1. 异常处理缺失：代码中的try块没有对应的catch或finally分支，无法处理future.get()抛出的InterruptedException、ExecutionException，以及getAllInputs方法可能抛出的异常。
2. 结果展示语法错误：
   • sout不是Java合法语法，需替换为System.out.println；
   • 使用stream().map()是错误的，map是中间操作不会触发执行，且此处不需要转换元素，应改用forEach；
   • Lambda表达式语法不完整，缺少分号和闭合括号。
3. 结果收集可优化：手动循环future.get()并逐个add元素到列表的写法冗余，可通过Stream API简化。

修正后的代码示例 #

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

// 假设inputs是已定义的输入数组，Info是自定义实体类，getAllInputs是已实现的Lambda API调用方法
List<Info> listOfInputs = new LinkedList<>();
List<CompletableFuture<List<Info>>> futures = new ArrayList<>();

try {
    // 批量拆分并提交并行任务
    for (int iterator = 0; iterator < inputs.length; iterator += 5) {
        int finalIterator = iterator;
        CompletableFuture<List<Info>> future = CompletableFuture.supplyAsync(() -> 
            getAllInputs(Arrays.copyOfRange(inputs, finalIterator, Math.min(inputs.length, finalIterator + 5)))
        );
        futures.add(future);
    }

    // 等待所有任务完成
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

    // 简化结果收集
    listOfInputs = futures.stream()
            .map(CompletableFuture::join) // 用join代替get，抛出unchecked异常
            .flatMap(List::stream)
            .collect(Collectors.toCollection(LinkedList::new));

    // 正确展示结果
    listOfInputs.forEach(infoEntry -> System.out.println("Hello" + infoEntry));
} catch (Exception e) {
    // 处理所有可能的异常，比如API调用失败、线程中断等
    e.printStackTrace();
}

额外优化建议 #

• 如果需要自定义线程池，可在supplyAsync中传入自定义Executor，避免占用默认ForkJoinPool的线程资源；
• 若对结果顺序无要求，当前逻辑已满足；若需保持输入的原始顺序，当前的批量拆分和收集逻辑也能保证顺序（因为循环提交任务的顺序和收集顺序一致）。

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