Java CompletableFuture并行处理大数据列表的优化实践

// 错误示例：导致串行执行
ExecutorService service = Executors.newFixedThreadPool(noOfCores - 1);
List<ResultBean> results = Lists.partition(largeList, 500).stream()
    .map(item -> CompletableFuture.supplyAsync(() -> executeListPart(item), service))
    .map(CompletableFuture::join) // 错误：在这里调用join会阻塞当前流的执行，直到当前Future完成
    .flatMap(List::stream)
    .collect(Collectors.toList());

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

// 假设的ListItem和ResultBean类
class ListItem {}
class ResultBean {}
class SomeService {
    public Optional<Object> methodA(ListItem item) {
        // 模拟耗时操作
        try { Thread.sleep(10); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return Optional.of(new Object());
    }
}

public class ParallelDataProcessor {

    private static SomeService service = new SomeService(); // 假设的服务实例

    // 假设的mapToBean方法
    private static ResultBean mapToBean(Object result, ListItem item) {
        // 实际的映射逻辑
        return new ResultBean();
    }

    // 模拟的executeListPart方法，它处理一个ListItem分区并返回List<ResultBean>
    private static List<ResultBean> executeListPart(List<ListItem> partition) {
        return partition.stream()
                .map(listItem -> service.methodA(listItem)
                        .map(result -> mapToBean(result, listItem)))
                .flatMap(Optional::stream)
                .collect(Collectors.toList());
    }

    public static void main(String[] args) throws InterruptedException {
        int noOfCores = Runtime.getRuntime().availableProcessProcessors();
        ExecutorService executor = Executors.newFixedThreadPool(noOfCores - 1);

        // 模拟一个大型列表
        List<ListItem> largeList = new java.util.ArrayList<>();
        for (int i = 0; i < 50000; i++) {
            largeList.add(new ListItem());
        }

        // 1. 将大型列表分区
        List<List<ListItem>> partitionedList = Lists.partition(largeList, 500);

        // 2. 提交异步任务并收集CompletableFuture实例
        List<CompletableFuture<List<ResultBean>>> futures = partitionedList.stream()
                .map(partition -> CompletableFuture.supplyAsync(() -> executeListPart(partition), executor))
                .collect(Collectors.toList());

        // ... 后续等待和结果收集
        // 3. 等待所有CompletableFuture完成并收集结果
        List<ResultBean> finalResults = futures.stream()
                .map(CompletableFuture::join) // 在所有Future都已提交后，统一等待并获取结果
                .flatMap(List::stream)      // 将List<List<ResultBean>>扁平化为List<ResultBean>
                .collect(Collectors.toList());

        System.out.println("Total processed items: " + finalResults.size());

        // 4. 关闭ExecutorService
        executor.shutdown();
        if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
            executor.shutdownNow();
        }
    }
}

// 承接上一步的代码
List<ResultBean> finalResults = futures.stream()
    .map(CompletableFuture::join) // 在所有Future都已提交后，统一等待并获取结果
    .flatMap(List::stream)      // 将List<List<ResultBean>>扁平化为List<ResultBean>
    .collect(Collectors.toList()); // 收集所有结果