处理大数组的多线程计算答案

【问题标题】：Handling multithreading computation of big array处理大数组的多线程计算
【发布时间】：2019-06-21 09:44:05
【问题描述】：

我有一个大数组，需要在该数组的每个元素上执行繁重的 CPU 工作。

基于my similar question，Alexei Kaigorodov 先生建议最好的方法是在每个单独的线程上拆分计算每个数据块。

还有我使用Kotlin协程实现算法的实现：

suspend fun predictAll(movingVehicles: List<MovingVehicle>): List<MovingVehicle?> {
    val prevTime = Timestamp(Date().time)
    val nextTime = Timestamp(Date().time)
    val ctx = Dispatchers.Default
    val processors = Runtime.getRuntime().availableProcessors()
    val chunks = movingVehicles.chunked(movingVehicles.count() / processors)
    val s = coroutineScope {
        val res = mutableListOf<Deferred<List<MovingVehicle?>>>()
        for (c in chunks) {
            val r = async(ctx) {
                c.map { predictLocation(it, prevTime, nextTime) }
            }
            res.add(r)
        }
        res.awaitAll()
    }
    return s.flatten()
}

private fun predictLocation(
    mv: MovingVehicle,
    prevTime: Timestamp,
    nextTime: Timestamp,
    relevance: Int = 5
): MovingVehicle?

它有效，但也许有更好的方法？我正在寻找 ExecutorService，但看起来它需要比协程更多的样板代码。

【问题讨论】：

我投票结束这个问题作为离题，因为这个社区不是为了审查工作代码（也许，只是也许：codereview.stackexchange.com 会是你请求的更好的地方）

标签： java multithreading kotlin kotlin-coroutines

【解决方案1】：

这实际上是一种使用协程的 Kotilinic 方式。您提交可以并发执行的异步任务，然后等待它们完成。

值得深思。一切都在一个线程中执行。这意味着协同程序也在线程上执行，如果您的任务阻塞线程将被阻塞。协程不会保存在那里。因此，创建一个 Threadpool 具有似乎最适合应用程序的属性（背压机制、最小/最大线程数等）通常是一个好主意。

现在，在您的情况下，您有 cpu-bound 任务，您无法通过拥有大量线程来获得更高的性能。对于此类任务，Amdahl's_law 的实际应用给出了 -

#threads = #cpu-cores - 1

默认情况下，协程由common pool 支持，这与上面提到的线程数相同，因此保留默认设置似乎很好。

但是，可能有多个库正在使用此池，并且如果您在其中任何一个库中有 IO 阻塞任务，您将失去性能。我建议您创建自己的 ForkJoinPool 并将其用作调度程序

val nOfThreads = Runtime.getRuntime().availableProcessors() - 1;
val ctx = ForkJoinPool( if (nOfThreads == 0) then 1 else nOfThreads).asCoroutineDispatcher()

【讨论】：