如何顺序运行Kotlin Coroutines？

Question

我有一个coroutinescope和log（）函数的实例，该函数看起来如下：

private val scope = CoroutineScope(Dispatchers.IO)

fun log(message: String) = scope.launch { // launching a coroutine
    println("$message")
    TimeUnit.MILLISECONDS.sleep(100) // some blocking operation
}

我使用此测试代码启动Coroutines：

repeat(5) { item ->
    log("Log $item")
}

log（）函数可以从任何位置，在任何thread中都从任何位置调用，但不能从coroutine中调用。

经过几次测试后，我可以看到如下的顺序结果：

Log 0
Log 2
Log 4
Log 1
Log 3

打印日志的顺序可能不同。如果我正确理解coroutines的执行不能保证是顺序的。这意味着可以在项目0的Coroutine之前启动项目2的Coroutine。

我希望为每个项目依次启动Coroutines，并且“某些阻止操作”将依次执行，以始终实现下一个日志：

Log 0
Log 1
Log 2
Log 3
Log 4

有没有办法使启动Coroutines顺序？也许还有其他方法可以实现我想要的东西？

在此先感谢您的任何帮助！

Answer 1

一种可能的策略是使用渠道按顺序加入启动的作业。您需要懒洋洋地启动工作，以便在加入之前不会启动。 Trysend在频道具有无限容量时始终成功。您需要使用trysend，因此可以从Coroutine外部调用。

private val lazyJobChannel = Channel<Job>(capacity = Channel.UNLIMITED).apply {
    scope.launch {
        consumeEach { it.join() }
    }
}

fun log(message: String) {
    lazyJobChannel.trySend(
        scope.launch(start = CoroutineStart.LAZY) {
            println("$message")
            TimeUnit.MILLISECONDS.sleep(100) // some blocking operation
        }
    )
}

Answer 2

因为 flow flow是顺序我们可以使用mutableSharedFlow依次收集和处理数据：

class Info {
    // make sure replay(in case some jobs were emitted before sharedFlow is being collected and could be lost)
    // and extraBufferCapacity are large enough to handle all the jobs. 
    // In case some jobs are lost try to increase either of the values.
    private val sharedFlow = MutableSharedFlow<String>(replay = 10, extraBufferCapacity = 10)
    private val scope = CoroutineScope(Dispatchers.IO)

    init {
        sharedFlow.onEach { message ->
            println("$message")
            TimeUnit.MILLISECONDS.sleep(100) // some blocking or suspend operation
        }.launchIn(scope)
    }

    fun log(message: String) {
        sharedFlow.tryEmit(message) 
    }
}

fun test() {

    val info = Info()

    repeat(10) { item ->
        info.log("Log $item")
    }
}

它始终以正确的顺序打印日志：

Log 0
Log 1
Log 2
...
Log 9

它适用于所有情况，但是需要确保设置了足够的元素以replay和extrabufferCapacity mutableSharedFlow的参数处理所有项目。

---

另一种方法是

使用dispatchers.io.limitedParallelism（1）作为coroutinescope的上下文。如果Coroutines不包含暂停功能的调用，并从相同的 thread ，例如主线程，它使得它们依次运行。因此，此解决方案仅与阻止（非代码>悬挂）操作启动>启动 coroutine builder：

private val scope = CoroutineScope(Dispatchers.IO.limitedParallelism(1))

fun log(message: String) = scope.launch { // launching a coroutine from the same Thread, e.g. Main Thread
    println("$message")
    TimeUnit.MILLISECONDS.sleep(100) // only blocking operation, not `suspend` operation
}

事实证明，单线程调度程序是FIFO执行者。因此，将coroutinescope执行到一个线程解决问题。