我使用 Java 8 并行流测试了该解决方案：
jobs.parallelStream().forEach { it.execute() }
我发现 CPU 利用率可靠地达到 100%。作为参考，我使用了这个计算工作：
class MyJob {
fun execute(): Double {
val rnd = ThreadLocalRandom.current()
var d = 1.0
(1..rnd.nextInt(1_000_000)).forEach { _ ->
d *= 1 + rnd.nextDouble(0.0000001)
}
return d
}
}
请注意，它的持续时间随机变化，从零到执行 100,000,000 次 FP 乘法所需的时间。
出于好奇，我还研究了您添加到问题中的代码作为适合您的解决方案。我发现它有很多问题，例如：
- 将所有结果累积到一个列表中，而不是在它们可用时对其进行处理
- 提交最后一个作业后立即关闭结果通道，而不是等待所有结果
我自己编写了一些代码，并添加了一些代码来对 Stream API 单线进行基准测试。这里是：
const val NUM_JOBS = 1000
val jobs = (0 until NUM_JOBS).map { MyJob() }
fun parallelStream(): Double =
jobs.parallelStream().map { it.execute() }.collect(summingDouble { it })
fun channels(): Double {
val resultChannel = Channel<Double>(UNLIMITED)
val mainComputeChannel = Channel<MyJob>()
val poolComputeChannels = (1..commonPool().parallelism).map { _ ->
GlobalScope.actor<MyJob>(Dispatchers.Default) {
for (job in channel) {
job.execute().also { resultChannel.send(it) }
}
}
}
val allComputeChannels = poolComputeChannels + mainComputeChannel
// Launch a coroutine that submits the jobs
GlobalScope.launch {
jobs.forEach { job ->
select {
allComputeChannels.forEach { chan ->
chan.onSend(job) {}
}
}
}
}
// Run the main loop which takes turns between running a job
// submitted to the main thread channel and receiving a result
return runBlocking {
var completedCount = 0
var sum = 0.0
while (completedCount < NUM_JOBS) {
select<Unit> {
mainComputeChannel.onReceive { job ->
job.execute().also { resultChannel.send(it) }
}
resultChannel.onReceive { result ->
sum += result
completedCount++
}
}
}
sum
}
}
fun main(args: Array<String>) {
measure("Parallel Stream", ::parallelStream)
measure("Channels", ::channels)
measure("Parallel Stream", ::parallelStream)
measure("Channels", ::channels)
}
fun measure(task: String, measuredCode: () -> Double) {
val block = { print(measuredCode().toString().substringBefore('.')) }
println("Warming up $task")
(1..20).forEach { _ -> block() }
println("\nMeasuring $task")
val average = (1..20).map { measureTimeMillis(block) }.average()
println("\n$task took $average ms")
}
这是我的典型结果：
Parallel Stream took 396.85 ms
Channels took 398.1 ms
结果差不多，但一行代码仍然胜过 50 行代码:)