兰特流的高性能缓冲

Question

我的代码会消耗大量（目前为数百万，最终为数十亿）相对较短（5-100 个元素）的随机数数组，并对它们进行一些不太费力的数学运算。随机数是随机数，理想情况下我想在多个内核上生成它们，因为随机数生成占我分析运行时间的 50% 以上。但是，我很难以不比单线程方法慢的方式分发大量小任务。

我的代码目前看起来像这样：

for(int i=0;i<1000000;i++){
    for(RealVector d:data){
        while(!converged){
            double[] shortVec = new double[5];
            for(int i=0;i<5;i++) shortVec[i]=rng.nextGaussian();
            double[] longerVec = new double[50];
            for(int i=0;i<50;i++) longerVec[i]=rng.nextGaussian();
            /*Do some relatively fast math*/
        }
    }
}

我采取的无效的方法是：

• 1+ 线程填充 ArrayBlockingQueue，而我的主循环消耗和填充数组（装箱/拆箱是这里的杀手）
• 在执行数学的非依赖部分时使用 Callable 生成向量（产生未来）（似乎间接的开销超过了我获得的任何并行性收益）
• 使用 2 个 ArrayBlockingQueue，每个由一个线程填充，一个用于短数组，一个用于长数组（仍然大约是直接单线程情况的两倍）。

我不是在寻找针对我的特定问题的“解决方案”，而是如何处理并行生成大量小型、独立原语并从单个线程中使用它们的一般情况。

Answer 1

这比使用队列更有效，因为；

• 有效负载是double[] 的数组，这意味着后台线程可以在传递数据之前生成更多数据。
• 所有对象都被回收。

.

public class RandomGenerator {
    private final ExecutorService generator = Executors.newSingleThreadExecutor(new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "generator");
            t.setDaemon(true);
            return t;
        }
    });
    private final Exchanger<double[][]> exchanger = new Exchanger<>();
    private double[][] buffer;
    private int nextRow = Integer.MAX_VALUE;

    public RandomGenerator(final int rows, final int columns) {
        buffer = new double[rows][columns];
        generator.submit(new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                Random random = new Random();
                double[][] buffer2 = new double[rows][columns];
                while (!Thread.interrupted()) {
                    for (int r = 0; r < rows; r++)
                        for (int c = 0; c < columns; c++)
                            buffer2[r][c] = random.nextGaussian();
                    buffer2 = exchanger.exchange(buffer2);
                }
                return null;
            }
        });
    }

    public double[] nextArray() throws InterruptedException {
        if (nextRow >= buffer.length) {
            buffer = exchanger.exchange(buffer);
            nextRow = 0;
        }
        return buffer[nextRow++];
    }
}

---

Random 是线程安全和同步的。这意味着每个线程都需要自己的 Random 来并发执行。

如何处理并行生成大量小型独立原语并从单个线程使用它们的一般情况。

我会使用Exchanger<double[][]> 在后台填充值以有效地传递它们（没有太多 GC 开销）

Answer 2

您的性能问题似乎是单个作业太小，因此大部分时间都花在了同步和排队作业本身上。要考虑的一件事是不是生成大量的小作业，而是向每个工作线程提供一个中等大小的作业集合，它将用答案进行注释。

例如，不是用第一个线程执行迭代 #0 来迭代循环，而是让下一个线程执行迭代 #1，... 我会让第一个线程执行迭代 #0 到 #999 或类似的。他们应该独立工作，并用他们的计算答案注释 Job 类。然后最后他们可以将已完成的整个作业集合返回为Future。

您的Job 类可能类似于以下内容：

public class Job {
    Collection<RealVector> dataCollection;
    Collection<SomeAnswer> answerCollection = new ArrayList<SomeAnswer>();
    public void run() {
        for (RealVector d : dataCollection) {
           // do the magic work on the vector
           while(!converged){
              ...
           }
           // put the associated "answer" in another collection
           answerCollection.add(someAnswer);
        }
    }
}