Java中如何多线程并行化超大区间的函数求和循环？

问题描述 #

给定整数a、b形成区间[a,b]，区间长度可超过10^9。需对区间内所有满足a ≤ n ≤ b的整数n，求函数f: N→N的取值之和。由于区间过大，希望通过多线程实现并行化，对应单线程代码如下：

long sum = 0;
    
for (long n = a ; n <= b ; n++)
    sum += f(n);
    
System.out.println(sum);

理想状态下，将区间按处理器可用线程数均分（假设每个n对应的f(n)复杂度与运行时间相近），各计算任务完全独立，f可为任意函数（例如求数字各位和）。请问在Java中是否有通用实现方式？

通用实现方案 #

1. 使用Stream API（Java 8+） #

Stream API自带并行流支持，底层会自动基于可用处理器数分配线程，代码简洁易维护：

import java.util.stream.LongStream;

public class ParallelSum {
    public static long parallelSum(long a, long b, java.util.function.LongUnaryOperator f) {
        return LongStream.rangeClosed(a, b)
                         .parallel()
                         .map(f)
                         .sum();
    }

    // 示例函数：计算数字各位和
    public static int digitSum(long n) {
        int sum = 0;
        n = Math.abs(n);
        while (n > 0) {
            sum += n % 10;
            n /= 10;
        }
        return sum;
    }

    public static void main(String[] args) {
        long a = 1;
        long b = 1000000000L;
        long result = parallelSum(a, b, ParallelSum::digitSum);
        System.out.println(result);
    }
}

• 优势：无需手动管理线程，代码简洁，自动适配处理器核心数
• 注意点：f必须是无副作用的纯函数，避免线程安全问题

2. 使用ExecutorService手动拆分区间 #

如果需要更精细地控制线程池或区间拆分逻辑，可以手动拆分区间后提交任务：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class ExecutorSum {
    public static long parallelSum(long a, long b, java.util.function.LongUnaryOperator f) {
        int threadCount = Runtime.getRuntime().availableProcessors();
        long interval = (b - a + 1) / threadCount;
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        List<Future<Long>> futures = new ArrayList<>();

        for (int i = 0; i < threadCount; i++) {
            long start = a + i * interval;
            long end = (i == threadCount - 1) ? b : start + interval - 1;
            // 提交子任务
            futures.add(executor.submit(() -> {
                long subSum = 0;
                for (long n = start; n <= end; n++) {
                    subSum += f.applyAsLong(n);
                }
                return subSum;
            }));
        }

        long totalSum = 0;
        try {
            for (Future<Long> future : futures) {
                totalSum += future.get();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executor.shutdown();
        }
        return totalSum;
    }

    // 示例函数：计算数字各位和
    public static int digitSum(long n) {
        int sum = 0;
        n = Math.abs(n);
        while (n > 0) {
            sum += n % 10;
            n /= 10;
        }
        return sum;
    }

    public static void main(String[] args) {
        long a = 1;
        long b = 1000000000L;
        long result = parallelSum(a, b, ExecutorSum::digitSum);
        System.out.println(result);
    }
}

• 优势：可自定义线程池参数、区间拆分规则，灵活性更高
• 注意点：需手动处理线程池关闭、异常捕获，确保所有任务完成后汇总结果

3. 使用Fork/Join框架（Java 7+） #

Fork/Join框架专门用于分治任务，适合递归拆分大任务：

import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;

public class ForkJoinSum extends RecursiveTask<Long> {
    private final long start;
    private final long end;
    private final java.util.function.LongUnaryOperator f;
    private static final long THRESHOLD = 1000000; // 阈值：当区间小于该值时直接计算

    public ForkJoinSum(long start, long end, java.util.function.LongUnaryOperator f) {
        this.start = start;
        this.end = end;
        this.f = f;
    }

    @Override
    protected Long compute() {
        long length = end - start + 1;
        if (length <= THRESHOLD) {
            // 直接计算子区间和
            long sum = 0;
            for (long n = start; n <= end; n++) {
                sum += f.applyAsLong(n);
            }
            return sum;
        } else {
            // 拆分任务
            long mid = start + (length / 2);
            ForkJoinSum leftTask = new ForkJoinSum(start, mid, f);
            ForkJoinSum rightTask = new ForkJoinSum(mid + 1, end, f);
            leftTask.fork(); // 异步执行左任务
            return rightTask.compute() + leftTask.join(); // 计算右任务并等待左任务结果
        }
    }

    public static long parallelSum(long a, long b, java.util.function.LongUnaryOperator f) {
        ForkJoinPool pool = new ForkJoinPool();
        return pool.invoke(new ForkJoinSum(a, b, f));
    }

    // 示例函数：计算数字各位和
    public static int digitSum(long n) {
        int sum = 0;
        n = Math.abs(n);
        while (n > 0) {
            sum += n % 10;
            n /= 10;
        }
        return sum;
    }

    public static void main(String[] args) {
        long a = 1;
        long b = 1000000000L;
        long result = parallelSum(a, b, ForkJoinSum::digitSum);
        System.out.println(result);
    }
}

• 优势：自动处理任务拆分与合并，适合大规模分治场景
• 注意点：需合理设置任务拆分阈值，避免过度拆分带来的性能开销

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