Kotlin：如何在没有 runBlocking 的情况下等待非挂起的协程？

Question

编辑 2： 我想我误解了文档。我读到：

运行阻塞

这个函数不应该在协程中使用。它旨在将常规阻塞代码连接到以挂起样式编写的库，用于main 函数和测试。

意味着我不应该使用runBlocking()，除了main 或测试。但我现在意识到我读错了，特别是这部分：

它旨在将常规阻塞代码连接到以挂起样式编写的库

所以看来runBlocking应该用在这种场景中。

但是我认为我应该完全理解上下文的主题，看看在这种情况下哪些上下文最适合传递给 runBlocking：

return runBlocking(???????) {
    job1.await() + job2.await()
}

---

编辑：显然，我对这个问题的表述很糟糕，因为所有回答它的尝试都错过了实际问题和我提出的限制。所以让我们尝试一种不同的方法......

这行得通：

fun doSomething(): Int {
    val job1 = GlobalScope.async { calculateSomething() }
    val job2 = GlobalScope.async { calculateSomething() }
    return runBlocking {
        job1.await() + job2.await()
    }
}

suspend fun calculateSomething(): Int {
    delay(1000L)
    return 13
}

suspend fun calculateSomethingElse(): Int {
    delay(2000L)
    return 19
}

我的问题是：无论如何我可以达到相同的结果：

1. 没有使用runBlocking()。
2. 没有将doSomething() 转换为suspend 函数。

?

换句话说：有什么我可以代替?????? 来完成以下工作的吗？

fun doSomething(): Int {
    val job1 = GlobalScope.async { calculateSomething() }
    val job2 = GlobalScope.async { calculateSomething() }
    return ????????? {
        job1.?????() + job2.?????()
    }
}

suspend fun calculateSomething(): Int {
    delay(1000L)
    return 13
}

suspend fun calculateSomethingElse(): Int {
    delay(2000L)
    return 19
}

---

我有一个小型实用程序方法，可以运行任何给定的外部命令并返回其输出（即 Java Process API 的小型包装器）：

class RunningCommand(private val proc: Process) {

    fun waitFor(): String {
        proc.outputStream.close()

        var output = ""
        val outputRdr = thread { output = proc.inputStream.bufferedReader().use { it.readText() } }
        var error = ""
        val errorRdr = thread { error = proc.errorStream.bufferedReader().use { it.readText() } }

        proc.waitFor()

        outputRdr.join()
        errorRdr.join()

        if (proc.exitValue() != 0) {
            throw RuntimeException("Command returned non-zero status: $output$error")
        }

        return output
    }
}

此代码可以正常工作。但是，它会为每个命令执行创建两个额外的线程。我想通过切换到协程来避免这种情况。我能够做到这一点，但我不得不使用runBlocking：

class RunningCommand(private val proc: Process) {

    fun waitFor(): String {
        proc.outputStream.close()

        var output = ""
        val outputRdr = GlobalScope.async { output = proc.inputStream.bufferedReader().use { it.readText() } }
        var error = ""
        val errorRdr = GlobalScope.async { error = proc.errorStream.bufferedReader().use { it.readText() } }

        proc.waitFor()

        runBlocking {
            outputRdr.await()
            errorRdr.await()
        }

        if (proc.exitValue() != 0) {
            throw RuntimeException("Command returned non-zero status: $output${error}")
        }

        return output
    }
}

这段代码也有效，但我读到runBlocking 应该只用于main() 方法和测试，即不应该以这种方式使用。看看它的实现，它看起来很糟糕，而且确实看起来像是不想从某个实用程序方法中重复调用的东西。

所以我的问题是：我应该如何弥合阻塞代码和协程之间的差距？或者换句话说，等待来自非suspend 代码的suspend 函数的正确方法是什么？

或者仅仅是我的设计是错误的，并且要在任何地方使用协程我需要创建 main() 方法 runBlocking 并且基本上总是在某个协程范围内？

Answer 1

您可以使用在后台执行操作的调度程序创建自己的范围。如果你想等待某事完全完成执行，你可以使用 withContext。

private val myScope = CoroutineScope(Dispatchers.Main) 

myScope.launch {

   withContext(Dispatchers.IO) {
        //to stuff in the background
    }

}

您运行下面的代码，您会看到它打印出 20，而不是 null。

fun main() {
    callSuspendFun()
}

suspend fun doWorkAndReturnDouble(num: Int): Int {
    delay(1000)
    return num * 2
}

fun callSuspendFun() {
    val coroutineScope = CoroutineScope(Dispatchers.Main)
    coroutineScope.launch {
        var num: Int? = null
        withContext(Dispatchers.IO) {
            val newNum = 10
            num = doWorkAndReturnDouble(newNum)
        }
        println(num)
    }
}

所以要从非挂起函数调用挂起函数，而不使用 runBlocking，你必须创建一个协程作用域。而 withContext 你等待代码的执行。

Answer 2

你应该使用coroutineScope

suspend fun waitFor(): String = coroutineScope {
    proc.outputStream.close()

    var output = ""
    val outputRdr = async { output = proc.inputStream.bufferedReader().use { it.readText() } }
    var error = ""
    val errorRdr = async { error = proc.errorStream.bufferedReader().use { it.readText() } }

    proc.waitFor()

        outputRdr.await()
        errorRdr.await()

    if (proc.exitValue() != 0) {
        throw RuntimeException("Command returned non-zero status: $output${error}")
    }

    return output
}

Answer 3

您可以将CoroutineScope() 与Dispathers.IO 一起使用，这将在后台线程中启动一个协程并卸载您在该线程上的执行

class RunningCommand(private val proc: Process) {

fun waitFor(): String {
    // Starting a new coroutine on Background thread (IO)
    proc.outputStream.close()
    var output = ""
    CoroutineScope(Dispatchers.Unconfined).async {

        val outputRdr = async { output = proc.inputStream.bufferedReader().use { it.readText() } }
        var error = ""
        val errorRdr = async { error = proc.errorStream.bufferedReader().use { it.readText() } }

        proc.waitFor()

        // No need of runBlocking 
        // await() call would wait for completion of its async block
        outputRdr.await() // Execution would block until outputRdr block completion on IO thread
        // At this stage, outputRdr block is completed
        errorRdr.await() // Execution would block until errorRdr block completion on IO thread
        // At this stage, errorRdr block is completed


        if (proc.exitValue() != 0) {
            throw RuntimeException("Command returned non-zero status: $output${error}")
        }
    return@async output
    }
    return output
}

}

注意：如果您从任何协程上下文调用waitFor() 方法，您可以通过编写coroutineScope { } 而不是CoroutineScope(Dispatchers.IO).launch { } 在相同的协程上下文上继续工作，从而正确管理结构化并发。

Answer 4

对于任何与我犯同样错误的未来旅行者 - runBlocking 不仅可以在 main / 测试中使用 - 而且：

它旨在将常规阻塞代码连接到以挂起样式编写的库

不知何故，我觉得只使用一些库函数是邪恶的，但事实并非如此。