Java中利用CompletableFuture高效并行处理大型列表数据

import com.google.common.collect.Lists;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

// 假设存在以下辅助类和方法
// class ListItem { /* ... */ }
// class ProcessResult { /* ... */ }
// class OutputBean { /* ... */ }
// class MyService { public Optional<ProcessResult> methodA(ListItem item) { /* ... */ } }
// class MyProcessor {
//     private MyService service = new MyService();
//     private OutputBean mapToBean(ProcessResult result, ListItem originalItem) { /* ... */ }
//     public List<OutputBean> executeListPart(List<ListItem> subList) {
//         return subList.stream()
//                       .map(listItem -> service.methodA(listItem)
//                                               .map(result -> mapToBean(result, listItem)))
//                       .flatMap(Optional::stream)
//                       .collect(Collectors.toList());
//     }
// }

public class ParallelProcessingIncorrect {

    // 假设这是您的列表和处理器实例
    private static List<ListItem> largeList = /* 初始化一个包含50k ListItem的列表 */;
    private static MyProcessor processor = new MyProcessor();

    public static void main(String[] args) {
        int noOfCores = Runtime.getRuntime().availableProcessors();
        ExecutorService service = Executors.newFixedThreadPool(noOfCores - 1);

        try {
            long startTime = System.currentTimeMillis();
            List<OutputBean> results = Lists.partition(largeList, 500).stream()
                    .map(item -> CompletableFuture.supplyAsync(() -> processor.executeListPart(item), service))
                    // 核心问题：在这里调用 CompletableFuture::join
                    .map(CompletableFuture::join)
                    .flatMap(List::stream)
                    .collect(Collectors.toList());
            long endTime = System.currentTimeMillis();
            System.out.println("Incorrect approach total time: " + (endTime - startTime) + " ms");
            System.out.println("Processed " + results.size() + " items.");

        } finally {
            service.shutdown();
        }
    }
}

import com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

// 辅助类定义（与上述示例相同，此处省略以保持简洁）
// class ListItem { /* ... */ }
// class ProcessResult { /* ... */ }
// class OutputBean { /* ... */ }
// class MyService { /* ... */ }
// class MyProcessor { /* ... */ }


public class ParallelProcessingCorrect {

    private static List<ListItem> largeList; // 假设已初始化，例如：
    static {
        largeList = new ArrayList<>();
        for (int i = 0; i < 50000; i++) {
            largeList.add(new ListItem("item_" + i));
        }
    }
    private static MyProcessor processor = new MyProcessor();

    public static void main(String[] args) throws InterruptedException {
        int noOfCores = Runtime.getRuntime().availableProcessors();
        ExecutorService service = Executors.newFixedThreadPool(noOfCores - 1); // 推荐线程池大小为核心数-1或根据IO/CPU密集型任务调整

        try {
            long startTime = System.currentTimeMillis();

            // 1. 创建并提交所有异步任务，收集CompletableFuture实例
            List<CompletableFuture<List<OutputBean>>> futures = Lists.partition(largeList, 500).stream()
                    .map(itemPart -> CompletableFuture.supplyAsync(() -> processor.executeListPart(itemPart), service))
                    .collect(Collectors.toList());

            // 2. 等待所有CompletableFuture完成并获取结果
            // 使用 CompletableFuture.allOf() 可以等待所有Future完成，但其本身不返回结果
            // 更好的做法是遍历futures列表，逐个join或使用allof().join()后，再map获取结果

            // 方法一：遍历futures列表，逐个join（更直接，但仍然是顺序join）
            // List<OutputBean> results = futures.stream()
            //                                  .map(CompletableFuture::join) // 此时join是等待所有任务提交后才开始
            //                                  .flatMap(List::stream)
            //                                  .collect(Collectors.toList());

            // 方法二：使用 CompletableFuture.allOf() 结合 thenApply/thenCompose（更优雅，推荐）
            CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

            List<OutputBean> results = allOf.thenApply(v -> futures.stream()
                                                .map(CompletableFuture::join) // 此时所有future都已完成，join是非阻塞的
                                                .flatMap(List::stream)
                                                .collect(Collectors.toList()))
                                        .join(); // 等待所有结果收集完成

            long endTime = System.currentTimeMillis();
            System.out.println("Correct approach total time: " + (endTime - startTime) + " ms");
            System.out.println("Processed " + results.size() + " items.");

        } finally {
            // 确保线程池关闭
            service.shutdown();
            if (!service.awaitTermination(60, TimeUnit.SECONDS)) {
                System.err.println("ExecutorService did not terminate in the specified time.");
                service.shutdownNow();
            }
        }
    }
}