最好的并行方式 [英] Best way to sum concurrently

问题描述

我想计算一些大数字。为了加快计算速度，我想利用多线程。

I am trying to compute some big numbers. In order to speed up computation, I would like to make use of multithreading. Each thread should calculate a number and in the end a sum is calculated.

我曾经看到过与 SumThread 和 Collector ，如下所示：

I once saw something that worked with a SumThread and a Collector that looked as follows:

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 而不是让新的线程不断变化：

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 的例子，但我需要一些加法器 BigInteger s ...

This code didn't manage to outperform the SumThread-Collector solution however. I've also seen things about LongAdder for instance, but I would need some adder for BigIntegers...

我的问题是：什么是同时计算总和的最好方法？

My question thus is: what is the best way to calculate a sum concurrently? Is it one of the above or is there a completely different (but better) way?

推荐答案

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

As you are mentioned LongAdder which was added in Java-8 and use effectively-final variables, I assume that you are using Java-8. In this version the best way to solve your task is to use the 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核心一样多。

Your problem is the classical map-reduce task where you should transform each element of some collection, then combine the results of the individual transformations into the final result. The Stream API is capable to parallelize such tasks quite effectively without any manual work. In Oracle JDK the tasks are executed in the common ForkJoinPool pool which by default creates as many threads as many CPU cores you have.

这篇关于最好的并行方式的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！