使用Java ForkJoin框架未实现并行性能提升的问题排查

结果异常的原因与澄清 #

你推测结果变化是整数溢出导致，但这个场景下并不存在溢出问题：1000万元素每个最大值为4，总和最大为40,000,000，远小于int类型的最大值2,147,483,647。如果结果偶尔变化，大概率是不同测试运行时数组由Random随机生成（默认用系统时间做种子），同一次运行中并行和串行使用的是同一个数组，结果应当一致。若确实出现同一次运行结果不同，可排查是否有测试环境的偶然干扰——你的并行代码中invokeAll会等待所有子任务完成后才累加结果，这部分逻辑是线程安全的。

并行性能更慢的核心原因 #

并行实现耗时更长，主要源于以下几点：

• 任务拆分开销过大：设置的THRESHOLD=1000会将任务拆分成数千个小任务，ForkJoinPool调度这些小任务的开销远大于并行计算带来的收益——求和是计算密度极低的操作，单个任务的计算时间远小于线程调度成本。
• JIT预热顺序问题：测试先运行并行代码再运行串行代码，JVM即时编译器（JIT）在第一次运行时未完成优化，并行代码处于解释执行状态；而串行运行时JIT已完成优化，因此速度更快。
• 冗余的ForkJoin调用：代码同时调用了forkJoinPool.execute(task)和forkJoinPool.invoke(task)，invoke本身就会提交任务并等待完成，execute属于多余操作，增加了不必要的开销。
• 缓存局部性损失：串行遍历数组时CPU缓存命中率极高（连续内存访问）；并行时多线程访问数组不同片段，会导致缓存行失效，降低缓存利用率。

优化方案 #

1. 保障结果准确性 #

将求和结果类型从int改为long，即使当前场景不会溢出，这也是避免潜在问题的良好编程习惯。

2. 提升并行性能 #

• 增大任务拆分阈值：将THRESHOLD调整为100,000甚至更高，减少任务数量，降低调度开销。
• 调整测试顺序并预热：先运行几次测试代码完成JIT预热，再执行正式测试；或调换并行与串行的测试顺序，确保两者都在JIT优化后运行。
• 简化ForkJoinPool使用：移除多余的execute调用，直接用invoke提交任务并等待结果。
• 使用Java Stream API：Java 8+的IntStream.parallel().sum()底层做了大量优化，比手动实现的ForkJoin任务更高效。

优化后的代码示例 #

方式一：优化手动ForkJoin实现 #

import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

public class App {
    public static void main(String[] args) throws Exception {
        int[] input = generateInput(10_000_000);
        
        // JIT预热
        seq(input);
        par(input);
        
        // 正式测试
        System.out.println("\n---正式测试---");
        par(input);
        seq(input);
    }

    private static void seq(int[] input) {
        long startTime = System.currentTimeMillis();

        long result = 0;
        for (int num : input) {
            result += num;
        }
        
        System.out.println("Sequential Result: " + result);
        System.out.println("Sequential time: " + (System.currentTimeMillis() - startTime) + "ms");
    }

    private static void par(int[] input) {
        int cores = Runtime.getRuntime().availableProcessors();
        System.out.println("Number of cores on this machine: " + cores);
        
        ForkJoinPool forkJoinPool = new ForkJoinPool(cores);

        long startTime = System.currentTimeMillis();
        RecursiveSumTask task = new RecursiveSumTask(input, 0, input.length);
        forkJoinPool.invoke(task);

        System.out.println("Parallel Result: " + task.getResult());
        System.out.println("Parallel time: " + (System.currentTimeMillis() - startTime) + "ms");
    }

    public static int[] generateInput(int size) {
        Random random = new Random(42); // 固定种子，确保每次测试数组一致
        int[] array = new int[size];
        for (int i = 0; i < size; i++) {
            array[i] = random.nextInt(5);
        }
        return array;
    }
}

class RecursiveSumTask extends RecursiveAction {
    private int[] input;
    private int start;
    private int end;
    private static final int THRESHOLD = 100_000;
    private long result;

    public RecursiveSumTask(int[] input, int start, int end) {
        this.input = input;
        this.start = start;
        this.end = end;
    }

    @Override
    protected void compute() {
        if (end - start > THRESHOLD) {
            int mid = (start + end) / 2;
            RecursiveSumTask left = new RecursiveSumTask(input, start, mid);
            RecursiveSumTask right = new RecursiveSumTask(input, mid, end);
            invokeAll(left, right);
            this.result = left.getResult() + right.getResult();
        } else {
            process();
        }
    }

    private void process() {
        long sum = 0;
        for (int i = start; i < end; i++) {
            sum += input[i];
        }
        this.result = sum;
    }

    public long getResult() {
        return this.result;
    }
}

方式二：使用Stream API（简洁高效） #

import java.util.Random;
import java.util.stream.IntStream;

public class StreamSumTest {
    public static void main(String[] args) {
        int[] input = generateInput(10_000_000);
        
        // JIT预热
        IntStream.of(input).sum();
        IntStream.of(input).parallel().sum();
        
        // 串行测试
        long start = System.currentTimeMillis();
        long seqResult = IntStream.of(input).sum();
        System.out.println("Sequential Result: " + seqResult);
        System.out.println("Sequential time: " + (System.currentTimeMillis() - start) + "ms");
        
        // 并行测试
        start = System.currentTimeMillis();
        long parResult = IntStream.of(input).parallel().sum();
        System.out.println("Parallel Result: " + parResult);
        System.out.println("Parallel time: " + (System.currentTimeMillis() - start) + "ms");
    }

    public static int[] generateInput(int size) {
        Random random = new Random(42);
        int[] array = new int[size];
        for (int i = 0; i < size; i++) {
            array[i] = random.nextInt(5);
        }
        return array;
    }
}

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