Java并行处理大型列表：使用CompletableFuture提升性能

list.stream()
    .map(listItem -> service.methodA(listItem).map(result -> mapToBean(result, listItem)))
    .flatMap(Optional::stream)
    .collect(Collectors.toList());

ExecutorService service = Executors.newFixedThreadPool(noOfCores - 1);
Lists.partition(list, 500).stream() // 将大列表分成小块
    .map(item -> CompletableFuture.supplyAsync(() -> executeListPart(item), service)) // 提交异步任务
    .map(CompletableFuture::join) // 立即等待每个任务完成
    .flatMap(List::stream)
    .collect(Collectors.toList());

import com.google.common.collect.Lists; // 假设使用Guava的Lists.partition
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

public class ParallelProcessingExample {

    // 假设这是您的业务逻辑方法，处理列表的一个分片并返回结果列表
    // executeListPart(List<MyItem> partition) 应该返回 List<MyProcessedBean>
    private List<MyProcessedBean> executeListPart(List<MyItem> partition) {
        // 模拟耗时操作
        try {
            Thread.sleep(100); // 假设每个分片处理100ms
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        // 实际业务逻辑：处理partition中的每个MyItem，并生成MyProcessedBean
        return partition.stream()
                .map(item -> new MyProcessedBean("Processed_" + item.getId())) // 示例转换
                .collect(Collectors.toList());
    }

    public List<MyProcessedBean> processLargeListInParallel(List<MyItem> largeList, int partitionSize, int threadPoolSize) {
        // 1. 创建并配置线程池
        // 建议线程池大小根据CPU核心数和任务类型（IO密集型/CPU密集型）调整
        ExecutorService executorService = Executors.newFixedThreadPool(threadPoolSize);

        try {
            // 2. 将大列表分成小块，并为每个小块创建异步任务
            // CompletableFuture<List<MyProcessedBean>> 表示每个任务会返回一个MyProcessedBean列表
            List<CompletableFuture<List<MyProcessedBean>>> futures = Lists.partition(largeList, partitionSize).stream()
                    .map(partition -> CompletableFuture.supplyAsync(() -> executeListPart(partition), executorService))
                    .collect(Collectors.toList());

            // 3. 创建一个CompletableFuture，等待所有子任务完成
            CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

            // 4. 当所有子任务完成后，聚合结果
            List<MyProcessedBean> finalResults = allOf.thenApply(v ->
                    futures.stream()
                            .map(CompletableFuture::join) // 此时所有future都已完成，join是非阻塞的
                            .flatMap(List::stream)       // 扁平化List<List<MyProcessedBean>>为List<MyProcessedBean>
                            .collect(Collectors.toList())
            ).join(); // 阻塞等待最终结果的聚合

            return finalResults;

        } finally {
            // 5. 关闭线程池，释放资源
            executorService.shutdown();
            // 可选：等待线程池终止，确保所有任务都已完成
            // try {
            //     if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
            //         executorService.shutdownNow();
            //     }
            // } catch (InterruptedException ex) {
            //     executorService.shutdownNow();
            //     Thread.currentThread().interrupt();
            // }
        }
    }

    // 示例数据类
    static class MyItem {
        private String id;
        public MyItem(String id) { this.id = id; }
        public String getId() { return id; }
    }

    static class MyProcessedBean {
        private String processedId;
        public MyProcessedBean(String processedId) { this.processedId = processedId; }
        public String getProcessedId() { return processedId; }
        @Override
        public String toString() { return "MyProcessedBean{" + "processedId='" + processedId + '\'' + '}'; }
    }

    public static void main(String[] args) {
        ParallelProcessingExample app = new ParallelProcessingExample();

        // 构造一个大型列表
        List<MyItem> largeList = new java.util.ArrayList<>();
        for (int i = 0; i < 5000; i++) {
            largeList.add(new MyItem("item_" + i));
        }

        long startTime = System.currentTimeMillis();
        List<MyProcessedBean> results = app.processLargeListInParallel(largeList, 500, Runtime.getRuntime().availableProcessors() - 1);
        long endTime = System.currentTimeMillis();

        System.out.println("Processed " + results.size() + " items in " + (endTime - startTime) + " ms");
        // System.out.println("First 10 results: " + results.subList(0, Math.min(10, results.size())));
    }
}