multithreading - Kotlin:使用非阻塞I/O阻塞协程

Question

我正在尝试使用Kotlin协程来处理非阻塞I / O。方案如下:

从线程1上运行的异步回调接收数据。

等待线程2中的此数据，然后使用它。

我当前的代码如下所示(为简洁起见简化):

private var latch = CountDownLatch(1)
private var data: Any? = null

// Async callback from non-blocking I/O
fun onReceive(data: Any) {
    currentData = data
    latch.countDown()
}

// Wait and consume data
fun getData(): Any? {
    latch.await()
    latch = CountDownLatch(1)
    return currentData
}

fun processData() {
    launch(CommonPool) {
        while (true) {
            val data = getData()
            // Consume data                
        }
    }
}

据我了解，Kotlin协程应该能够帮助我摆脱CountDownLatch。阅读 this (awesome) guide之后，我所能想到的就是这样:

// Wait and consume data
fun getData() = async(CommonPool) {
    latch.await()
    latch = CountDownLatch(1)
    currentData
}

fun processData() {
    launch(CommonPool) {
        while (true) {
            runBlocking {
                val data = getData().await()
                // Consume data                
            }
        }
    }
}

我还尝试了 Pipelines，结果相似。我显然不了解如何使用这些功能。

Answer 1

您没有说onReceive()中接收的数据是否可以并行处理。这是主要问题。如果是，则可以使用onReceive()进行操作。如果不允许这样做，则让每个对onReceive()的调用都在CommonPool上启动一个没有任何协程的任务。如果应该顺序处理它们，那么最简单的方法是使用内部循环来启动线程:

fun onReceive(data: Any) {
   queue.put(data);
}

 ....

// loop in a thread
while(true) {
   data = queue.take();
   processData(data);
}

同样，不需要协程。

通常，协程是语法糖，用来表示异步程序，就好像它是同步的一样。我不认为您的程序是使用协程的案例。