挂起 IO 函数永不返回

Question

我在为 Android 编写 UDP 消息接收循环时遇到了困难。

在下面的代码中，在receiveLoop 中，对receiveMessages 的调用永远不会返回，因此我永远不会进入消息处理循环。 请注意，我仍然可以接收数据包，但是当通道缓冲区已满时它会停止。

我希望receiveMessages 立即返回，而其中的阻塞 IO 循环仍将永远运行。

class MySocketUDP(private val params: SocketParams) {


    private val rcvSocket: DatagramSocket by lazy {
        val sock = DatagramSocket(params.rcvPort)
        sock.reuseAddress = true
        sock.soTimeout = 1000
        sock
    }

    suspend fun receiveMessages(channel: SendChannel<Message>) {
        withContext(Dispatchers.IO) {
            val buf = ByteArray(MAX_MSG_SIZE)
            while (true) {
                val pkt = DatagramPacket(buf, buf.size)
                try {
                    if (channel.isClosedForSend) {
                        break
                    }
                    rcvSocket.receive(pkt)
                    val msg = packetToMessage(buf, 0, pkt.length)
                    Log.d("SOCKET", "filling channel with $msg")
                    channel.send(msg)
                } catch (ex: SocketTimeoutException) {
                } catch (ex: CancellationException) {
                    break
                }
            }
        }
    }
}

class MyModel {

    private suspend fun receiveLoop(socket: MySocketUDP) {
        withContext(Dispatchers.Main) {
            val channel = Channel<Message>(16)
            socket.receiveMessages(channel)
            Log.d("MODEL", "Entering msg loop")
            for (msg in channel) {
                dispatchRcvMessage(msg)
            }
        }
    }

}

1. 为什么receiveMessages 在IO 调度程序中运行并从Main 调度程序调用时从不返回？
2. 我是否需要为此类生产者/消费者工作实际生成一个线程？
3. 您能否展示如何以“协程友好”的方式很好地实现如此长的阻塞代码？

谢谢

Answer 1

1. receiveMessages() 是一个挂起函数，它调用另一个挂起函数withContext()，这又是一个无限循环。所以调用socket.receiveMessages(channel)会在循环未完成时暂停代码执行。

2. 您需要为消费者和生产者启动单独的协程，例如使用launch 函数。

3. 使用协程的一些例子：

    val someScope = CoroutineScope(Dispatchers.Main)
   
    private suspend fun receiveLoop(socket: MySocketUDP) = someScope.launch {
        val channel = Channel<Message>(16)
        socket.receiveMessages(channel)
        Log.d("MODEL", "Entering msg loop")
        for (msg in channel) {
            dispatchRcvMessage(msg)
        }
    }
   
    // In MySocketUDP
    suspend fun receiveMessages(channel: SendChannel<Message>) {
        someAnotherScope.launch { // or can use coroutineScope builder function
            val buf = ByteArray(MAX_MSG_SIZE)
            while (true) {
                val pkt = DatagramPacket(buf, buf.size)
                try {
                    if (channel.isClosedForSend) {
                        break
                    }
                    rcvSocket.receive(pkt)
                    val msg = packetToMessage(buf, 0, pkt.length)
                    Log.d("SOCKET", "filling channel with $msg")
                    channel.send(msg)
                } catch (ex: SocketTimeoutException) {
                } catch (ex: CancellationException) {
                    break
                }
            }
        }
    }