在 Scala 中，可以轻松地做一个并行映射、forEach 等，使用：
collection.par.map(..)
在 Kotlin 中是否有等价物？
【问题讨论】：
标签： parallel-processing kotlin
Kotlin 标准库不支持并行操作。不过，由于 Kotlin 使用标准的 Java 集合类，您也可以使用 Java 8 流 API 对 Kotlin 集合执行并行操作。
例如
myCollection.parallelStream()
.map { ... }
.filter { ... }
【讨论】：
myCollection.parallelStream().map { ... }. filter { ... }
从 Kotlin 1.1 开始，并行操作也可以用 coroutines 非常优雅地表示。这是列表中的pmap：
fun <A, B>List<A>.pmap(f: suspend (A) -> B): List<B> = runBlocking {
map { async(CommonPool) { f(it) } }.map { it.await() }
}
请注意，协程仍是一个实验性功能。
【讨论】：
foo.pmap { v -> ... } 一起使用。我认为这相当优雅。
在 Kotlin 的标准库中还没有官方支持，但是你可以定义一个 extension function 来模仿 par.map：
fun <T, R> Iterable<T>.pmap(
numThreads: Int = Runtime.getRuntime().availableProcessors() - 2,
exec: ExecutorService = Executors.newFixedThreadPool(numThreads),
transform: (T) -> R): List<R> {
// default size is just an inlined version of kotlin.collections.collectionSizeOrDefault
val defaultSize = if (this is Collection<*>) this.size else 10
val destination = Collections.synchronizedList(ArrayList<R>(defaultSize))
for (item in this) {
exec.submit { destination.add(transform(item)) }
}
exec.shutdown()
exec.awaitTermination(1, TimeUnit.DAYS)
return ArrayList<R>(destination)
}
这是一个简单的使用示例
val result = listOf("foo", "bar").pmap { it+"!" }.filter { it.contains("bar") }
如果需要，它允许通过提供线程数甚至特定的java.util.concurrent.Executor 来调整线程。例如
listOf("foo", "bar").pmap(4, transform = { it + "!" })
请注意，这种方法只允许并行化map 操作，不会影响任何下游位。例如。第一个示例中的filter 将运行单线程。然而，在许多情况下，只有数据转换（即map）需要并行化。此外，将上述方法扩展到 Kotlin 集合 API 的其他元素也很简单。
【讨论】：
Sequence<R>（或Flow<R>）的版本感兴趣。不幸的是，我不能简单地让整个代码在= execute{ 块中执行并调用yield 而不是destination.add，因为yield 只能在原始块中执行，所以在exec.submit { } 内不是一个选项。 （无需保留订单。）
你可以使用这个扩展方法：
suspend fun <A, B> Iterable<A>.pmap(f: suspend (A) -> B): List<B> = coroutineScope {
map { async { f(it) } }.awaitAll()
}
更多信息请见Parallel Map in Kotlin
【讨论】：
从 1.2 版本开始，kotlin 添加了一个符合 JRE8 的stream feature
因此，异步迭代列表可以像下面这样完成：
fun main(args: Array<String>) {
val c = listOf("toto", "tata", "tutu")
c.parallelStream().forEach { println(it) }
}
【讨论】：
Kotlin 想要是惯用的，但又不至于过于合成，乍一看难以理解。
通过协程进行并行计算也不例外。他们希望通过一些预先构建的方法使其简单但不隐含，允许在需要时分支计算。
在你的情况下：
collection.map {
async{ produceWith(it) }
}
.forEach {
consume(it.await())
}
请注意，要调用async 和await，您需要位于所谓的Context 中，您不能在非协程上下文中进行挂起调用或启动协程。要输入一个，您可以：
runBlocking { /* your code here */ }：它将暂停当前线程，直到 lambda 返回。GlobalScope.launch { }：它将并行执行 lambda；如果您的main 执行完毕，而您的协程还没有发生坏事，那么最好使用runBlocking。希望对你有帮助:)
【讨论】：
我发现另一种非常优雅的方法是这样的，使用kotlinx.coroutines 库：
import kotlinx.coroutines.flow.asFlow
suspend fun process(myCollection: Iterable<Foo>) {
myCollection.asFlow()
.map { /* ... */ }
.filter { /* ... */ }
.collect { /* ... perform some side effect ... */ }
}
然而，它确实需要额外的依赖； kotlinx.coroutines 不在标准库中。
【讨论】：
目前没有。 Kotlin 与 Scala 的官方比较提到：
以后可能会添加到 Kotlin 中的东西：
- 并行集合
【讨论】：
此解决方案假定您的项目正在使用协程：
implementation( "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.2")
名为parallelTransform 的函数不保留元素的顺序并返回Flow<R>，而函数parallelMap 保留元素的顺序并返回List<R>。
为多个调用创建一个线程池：
val numberOfCores = Runtime.getRuntime().availableProcessors()
val executorDispatcher: ExecutorCoroutineDispatcher =
Executors.newFixedThreadPool(numberOfCores ).asCoroutineDispatcher()
使用该调度程序（并在不再需要时调用close()）：
inline fun <T, R> Iterable<T>.parallelTransform(
dispatcher: ExecutorDispatcher,
crossinline transform: (T) -> R
): Flow<R> = channelFlow {
val items: Iterable<T> = this@parallelTransform
val channelFlowScope: ProducerScope<R> = this@channelFlow
launch(dispatcher) {
items.forEach {item ->
launch {
channelFlowScope.send(transform(item))
}
}
}
}
如果线程池重用不重要（线程池并不便宜），您可以使用这个版本：
inline fun <T, R> Iterable<T>.parallelTransform(
numberOfThreads: Int,
crossinline transform: (T) -> R
): Flow<R> = channelFlow {
val items: Iterable<T> = this@parallelTransform
val channelFlowScope: ProducerScope<R> = this@channelFlow
Executors.newFixedThreadPool(numberOfThreads).asCoroutineDispatcher().use { dispatcher ->
launch( dispatcher ) {
items.forEach { item ->
launch {
channelFlowScope.send(transform(item))
}
}
}
}
}
如果您需要保留元素顺序的版本：
inline fun <T, R> Iterable<T>.parallelMap(
dispatcher: ExecutorDispatcher,
crossinline transform: (T) -> R
): List<R> = runBlocking {
val items: Iterable<T> = this@parallelMap
val result = ConcurrentSkipListMap<Int, R>()
launch(dispatcher) {
items.withIndex().forEach {(index, item) ->
launch {
result[index] = transform(item)
}
}
}
// ConcurrentSkipListMap is a SortedMap
// so the values will be in the right order
result.values.toList()
}
【讨论】：
我发现了这个：
实现'com.github.cvb941:kotlin-parallel-operations:1.3'
详情：
【讨论】：
我想出了几个扩展函数：
Iterable<T> 类型上的 suspend 扩展函数，它对项目进行并行处理并返回处理每个项目的一些结果。默认情况下，它使用Dispatchers.IO 调度程序将阻塞任务卸载到共享线程池。必须从协程（包括带有 Dispatchers.Main 调度程序的协程）或另一个 suspend 函数调用。
suspend fun <T, R> Iterable<T>.processInParallel(
dispatcher: CoroutineDispatcher = Dispatchers.IO,
processBlock: suspend (v: T) -> R,
): List<R> = coroutineScope { // or supervisorScope
map {
async(dispatcher) { processBlock(it) }
}.awaitAll()
}
协程调用示例：
val collection = listOf("A", "B", "C", "D", "E")
someCoroutineScope.launch {
val results = collection.processInParallel {
process(it)
}
// use processing results
}
其中someCoroutineScope 是CoroutineScope 的一个实例。
在CoroutineScope 上启动并忘记扩展功能，它不会返回任何结果。它还默认使用Dispatchers.IO 调度程序。可以使用CoroutineScope 或从另一个协程调用。
fun <T> CoroutineScope.processInParallelAndForget(
iterable: Iterable<T>,
dispatcher: CoroutineDispatcher = Dispatchers.IO,
processBlock: suspend (v: T) -> Unit
) = iterable.forEach {
launch(dispatcher) { processBlock(it) }
}
调用示例：
someoroutineScope.processInParallelAndForget(collection) {
process(it)
}
// OR from another coroutine:
someCoroutineScope.launch {
processInParallelAndForget(collection) {
process(it)
}
}
2a。在Iterable<T> 上启动并忘记扩展功能。和之前的差不多，只是扩展类型不同。 CoroutineScope 必须作为参数传递给函数。
fun <T> Iterable<T>.processInParallelAndForget(
scope: CoroutineScope,
dispatcher: CoroutineDispatcher = Dispatchers.IO,
processBlock: suspend (v: T) -> Unit
) = forEach {
scope.launch(dispatcher) { processBlock(it) }
}
调用：
collection.processInParallelAndForget(someCoroutineScope) {
process(it)
}
// OR from another coroutine:
someScope.launch {
collection.processInParallelAndForget(this) {
process(it)
}
}
【讨论】：