Java并行批量处理实现：如何并行计算数组各分区的求和结果

你的现有代码实际上已经实现了3个子数组的并行求和，你存在的误解是：遍历提交任务、遍历获取结果的过程不代表计算是串行的。所有求和任务在调用completionService.submit()之后就已经被线程池调度并行执行，后续的循环只是异步获取已完成的计算结果，不会影响任务的并行性。

下面给你两种更简洁的实现方案：

方案1：并行流实现（适合简单计算场景） #

无需手动维护线程池，代码量极少：

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

public class ParallelSum {
    public static void main(String[] args) {
        List<Integer> values = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12);
        // 分区+并行求和逻辑
        List<Integer> partitionSums = values.parallelStream()
                .collect(Collectors.groupingBy(index -> values.indexOf(index) / 4))
                .values()
                .parallelStream()
                .map(partition -> partition.stream().mapToInt(Integer::intValue).sum())
                .collect(Collectors.toList());
        // 打印结果
        partitionSums.forEach(System.out::println);
    }
}

方案2：CompletableFuture实现（适合复杂异步场景） #

支持自定义线程池、异常处理、结果编排，灵活度更高：

import com.google.common.collect.Lists;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CompletableFutureSum {
    public static void main(String[] args) {
        List<Integer> values = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12);
        List<List<Integer>> partitions = Lists.partition(values, 4);
        // 线程池核心数和分区数一致即可，避免资源浪费
        ExecutorService pool = Executors.newFixedThreadPool(3);

        // 批量提交所有并行求和任务，完成后自动打印结果
        CompletableFuture<Void> allTask = CompletableFuture.allOf(
                partitions.stream()
                        .map(partition -> CompletableFuture.supplyAsync(
                                () -> partition.stream().mapToInt(Integer::intValue).sum(), pool
                        ).thenAccept(System.out::println))
                        .toArray(CompletableFuture[]::new)
        );

        // 等待所有任务执行完成后关闭线程池
        allTask.join();
        pool.shutdown();
    }
}

额外优化建议 #

• 你的原有代码中线程池核心数设置为10，但实际仅需要处理3个任务，调整为和分区数一致即可，避免不必要的资源浪费
• 原有代码中future = completionService.take()覆盖了列表中的原有Future对象，没有必要，直接调用take()拿结果即可，不需要额外用列表存储Future

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