同时求和的最佳方法答案

【问题标题】：Best way to sum concurrently同时求和的最佳方法
【发布时间】：2015-11-15 01:15:33
【问题描述】：

我正在尝试计算一些大数字。为了加快计算速度，我想使用多线程。每个线程应该计算一个数字，最后计算一个总和。

我曾经看到与SumThread 和Collector 一起使用的东西，如下所示：

public BigInteger compute(int p) {
    Collector c = new Collector(p);

    for(T element : Collection<T> bigCollection) {
        new SumThread(c) {

            @Override
            protected void doTheJob() {
                long big = someVeryComplexCalculation(element, ...); //n!
                receive(BigInteger.valueOf(big));
            }

        }
    }

    if(collector.isReady())
        return collector.getResult();

    return null;
}

public class Collector {

    private int numberOfProcesses;
    private int numberOfAllowedProcesses;
    private BigInteger result;

    public Collector(int n) {
        numberOfAllowedProcesses = n;
        numberOfProcesses = 0;
        result = BigInteger.ZERO;
    }

    synchronized public void enter() throws InterruptedException {
        if (numberOfProcesses == numberOfAllowedProcesses) wait();
        numberOfProcesses++;
    }

    synchronized public void leave() {
        numberOfProcesses--;
        notify();
    }

    synchronized public void register(BigInteger v) {
        result = result.add(v);
    }

    synchronized public boolean isReady() throws InterruptedException {
        while (numberOfProcesses > 0) wait();
        return true;
    }

    ...
}

public abstract class SumThread extends Thread {

    private Collector collector;

    public SumThread(Collector c) throws InterruptedException {
        collector = c;
        collector.enter();
    }

    abstract protected void doTheJob(); //complex calculations can be done in here

    public void receive(BigInteger t) {
        collector.register(t);
    }

    public void run() {
        doTheJob();
        collector.leave();
    }
}

我认为我可以通过使用ExecutorService 而不是不断创建新的Threads 来轻松超越这一点：

public BigInteger compute(int p) {
    ExecutorService pool = Executors.newFixedThreadPool(p);
    Future<BigInteger>[] futures = new Future<BigInteger>[bigCollection.size()];
    int i = 0;

    for(T element : Collection<T> bigCollection) {
        futures[i++] = p.submit(new Callable<BigInteger>() {

            @Override
            public BigInteger call() {
                long big = someVeryComplexCalculation(element, ...); //n!
                return BigInteger.valueOf(big);
            }

        }
    }

    // or with ExecutorCompletionService, but the loop remains I guess
    BigInteger res = BigInteger.ZERO
    for(Future<BigInteger> f : futures)
        res = res.add(f.get());

    return res;
}

但是，此代码的性能并没有优于 SumThread-Collector 解决方案。例如，我也看到了关于 LongAdder 的事情，但我需要一些 BigIntegers 的加法器......

因此，我的问题是：同时计算总和的最佳方法是什么？是上述方法之一还是有完全不同（但更好）的方法？

【问题讨论】：

标签： java multithreading concurrency parallel-processing biginteger

【解决方案1】：

正如您提到的 LongAdder 是在 Java-8 中添加并使用有效最终变量的，我假设您使用的是 Java-8。在此版本中，解决您的任务的最佳方法是使用Stream API：

BigInteger result = bigCollection.parallelStream()
                     .map(e -> BigInteger.valueOf(someVeryComplexCalculation(e, ...)))
                     .reduce(BigInteger.ZERO, BigInteger::add);

您的问题是经典的 map-reduce 任务，您应该转换某个集合的每个元素，然后将各个转换的结果组合成最终结果。 Stream API 能够非常有效地并行化此类任务，而无需任何手动工作。在 Oracle JDK 中，任务在 common ForkJoinPool pool 中执行，默认情况下，它会创建与您拥有的 CPU 内核一样多的线程。

【讨论】：

这个确实很有效！只是出于好奇：例如，有没有可能将此应用于Iterable<T> 的实例？有没有办法压缩流？（例如：complexCalculation() = part1()*part2() 因此我想将 part1() 映射到 bigCollection 和 part2() 上，并通过乘法将两个结果流的结果并行合并）跨度>
@MrTsjolder，为了并行处理流，您的流源（可迭代或其他）必须提供一个可以很好地拆分源的拆分器。如果您怀疑您的来源，您可以将Iterable<T> 的内容复制到ArrayList<T>。至于压缩，如果没有看到确切的源代码，我无法回答。你可能会问一个不同的问题，提供所有必要的细节（使用“java-stream”标签，我总是监控它）。

【解决方案2】：

你有两个解决方案：

首先，我建议使用 JDK7 中的 Fork-Join 框架来完成此任务：

你需要实现一个 RecursiveTask

正如@tagir-valeev 所提议的那样，作为第二种解决方案 (JDK8) 将使用 pararell 流。

在这两种情况下，这取决于您的用途以及您使用的 Java 版本。

【讨论】：