Kotlin 协程进度计数器

Question

我正在使用 async/await 发出数千个 HTTP 请求，并且希望有一个进度指示器。我以一种天真的方式添加了一个，但注意到当所有请求完成时，计数器值永远不会达到总数。所以我创建了一个简单的测试，果然，它没有按预期工作:

fun main(args: Array<String>) {
    var i = 0
    val range = (1..100000)
    range.map {
        launch {
            ++i
        }
    }
    println("$i ${range.count()}")
}

输出是这样的，其中第一个数字总是变化:

98800 100000

我可能错过了有关 JVM/Kotlin 中并发/同步的一些重要细节，但不知道从哪里开始。有什么建议吗？

更新:我最终按照 Marko 的建议使用了 channel :

/**
 * Asynchronously fetches stats for all symbols and sends a total number of requests
 * to the `counter` channel each time a request completes. For example:
 *
 *     val counterActor = actor<Int>(UI) {
 *         var counter = 0
 *         for (total in channel) {
 *             progressLabel.text = "${++counter} / $total"
 *         }
 *     }
 */
suspend fun getAssetStatsWithProgress(counter: SendChannel<Int>): Map<String, AssetStats> {
    val symbolMap = getSymbols()?.let { it.map { it.symbol to it }.toMap() } ?: emptyMap()
    val total = symbolMap.size
    return symbolMap.map { async { getAssetStats(it.key) } }
        .mapNotNull { it.await().also { counter.send(total) } }
        .map { it.symbol to it }
        .toMap()
}

Answer 1

到底是什么导致你的错误方法失败的解释是次要的:首要的是修复方法。

对于这种通信模式，您应该拥有一个 actor，而不是 async-await 或 launch，所有 HTTP 作业都将其发送到该 actor地位。这将自动处理您的所有并发问题。

这是一些示例代码，取自您在评论中提供的链接，并根据您的用例进行了调整。参与者不是由第三方询问计数器值并用它更新 GUI，而是在 UI 上下文中运行并更新 GUI 本身:

import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.channels.*
import kotlin.system.*
import kotlin.coroutines.experimental.*

object IncCounter

fun counterActor() = actor<IncCounter>(UI) {
    var counter = 0
    for (msg in channel) {
        updateView(++counter)
    }
}

fun main(args: Array<String>) = runBlocking {
    val counter = counterActor()
    massiveRun(CommonPool) {
        counter.send(IncCounter)
    }
    counter.close()
    println("View state: $viewState")
}


// Everything below is mock code that supports the example
// code above:

val UI = newSingleThreadContext("UI")

fun updateView(newVal: Int) {
    viewState = newVal
}

var viewState = 0

suspend fun massiveRun(context: CoroutineContext, action: suspend () -> Unit) {
    val numCoroutines = 1000
    val repeatActionCount = 1000
    val time = measureTimeMillis {
        val jobs = List(numCoroutines) {
            launch(context) {
                repeat(repeatActionCount) { action() }
            }
        }
        jobs.forEach { it.join() }
    }
    println("Completed ${numCoroutines * repeatActionCount} actions in $time ms")
}

运行它打印

Completed 1000000 actions in 2189 ms
View state: 1000000