使用 Java 多线程，协调寻找最佳结果的最有效方法是什么？

Question

让我说清楚，我在下面描述的方法是可操作的。我希望提高该方法的吞吐量。它有效，而且效果很好。我们正在寻求进一步扩展吞吐量，这就是我正在研究这一点的原因。

手头的任务是提高评分算法的性能，该算法返回一组任务的最佳分数。我收集了使用ExecutorService. 每个任务检查它现在是否有最好的分数，如果它是新的最好的，则以同步的方式更新最好的分数。为了深入了解我正在处理的规模，每项任务只需要几分之一毫秒即可完成，但有数千个任务，因此需要数百毫秒才能找到最佳任务。我每分钟执行数百次这个评分算法。结果是 60 秒中有 30 秒用于运行此评分算法。

当我的线程池为 8 个线程（具有 24 个虚拟内核）时，每个任务需要 0.3 毫秒。当我有 20 个线程（同一台机器，24 个虚拟核心）时，每个任务需要 0.6 毫秒。我怀疑当我向我的ExecutorService线程池添加更多线程时，我的性能会因为最佳分数的同步而变得更糟（更多线程争用锁）。

我已经做了很多搜索，但似乎无法找到令人满意的（实际上，我似乎找不到任何）替代方案。我正在考虑收集所有分数并按排序顺序存储，或者在所有任务完成后排序——但我不确定这是否会有任何改进。

有没有人对另一种更有效的收集最高分的方法有任何想法？

这是当前的方法：

final double[] bestScore = { Double.MAX_VALUE };
// for each item in the collection {
    tasks.add(Executors.callable(new Runnable() {
        public void run() {
            double score = //... do the scoring for the task
            if (score < bestScore[0]) {
                synchronized(bestScore) {
                    if (score < bestScore[0]) { // check again after we have the lock
                        bestScore[0] = score;
                        ...
                        // also save off other task identifiers in a similar fashion
                    }
                }
            }
        }
    }
} // end of loop creating scoring tasks

List<Future<Object>> futures = executorService.invokeAll(tasks /*...timeout params here*/);
... // handle cancelled tasks 

// now use the best scoring task that was saved off when it was found.

Answer 1

我必须理所当然地认为您希望将每个单独的分数作为提交给ExecutorService. 一定还有其他好处，否则这些开销就不值得。通常，您会实现Callable在执行时返回分数（或具有分数和其他相关结果的对象）。成功调用所有任务后，将在主线程中检查所有结果以获得最佳结果。

然而，考虑到您的限制，您可以尝试的一种优化是使用 a DoubleAccumulator，它适用于此类情况，而不是单元素数组和同步。它看起来像这样：

final DoubleAccumulator lowest = new DoubleAccumulator(Math::min, Double.POSITIVE_INFINITY);
/* Loop, creating all the tasks... */
for ( ... ) {
  tasks.add(Executors.callable(new Runnable() {
    public void run()
    {
      double score = 0; /* Compute a real score here. */
      lowest.accumulate(score);
    }
  }));
}
/* Invoke all the tasks, when successful... */
double lowestScore = lowest.get();

如果您需要跟踪分数之外的信息，您可以执行类似的操作AtomicReference，创建一个包含任务标识符、分数和任何其他所需属性的数据对象，并使用其累加器之一。

如果您的任务是通过某种递归、分而治之的方法初始化的，从而产生非阻塞、大小相等的任务，那么并行底层的 fork-join 框架也Stream可能很适合。

不过，我再次指出，如果更多线程会降低性能，那么衡量更少线程的使用似乎是谨慎的做法。