如何在 Java 代码中使用 Kotlin 协程实现 NIO Socket（客户端）？

Question

我想用 Kotlin(v1.3.0) coroutines & java.nio.channels.SocketChannel (NIO) 替换 Socket connect (blocking IO) in安卓。因为这样可以节省很多线程。

下面的代码无法运行，因为job.await() 在 Kotlin 中是挂起函数，它只能在 Ktolin 协程块中调用。比如launch{..}、async{..}。

// this function will be called by Java Code
fun connect(address: InetSocketAddress, connectTimeout: Int): SocketChannel {

    // Start a new connection
    // Create a non-blocking socket channel
    val socketChannel = SocketChannel.open()
    socketChannel.configureBlocking(false)

    // async calls NIO connect function
    val job = GlobalScope.async(oneThreadCtx) {
        aConnect(socketChannel, address)
    }

    // I what to suspend(NOT block) current Java Thread, until connect is success
    job.await()

    return socketChannel
}

但是，我尝试使用runBlocking{..} 使这个函数成为Java 中的普通函数。但是job.await 阻止了当前的 Java 线程，不暂停。

那么，我应该如何使用 Kotlin(v1.3.0) 协程实现这个功能？

Answer 1

正如 Marko 指出的那样，即使该阻塞操作在异步协程中，您的代码仍将最终阻塞线程。要使用 Java 和 Kotlin 真正获得所需的异步行为，您需要使用 Socket Channel 的异步版本

这样，您可以获得真正的异步套接字处理。使用该类和 Kotlin 的 suspendCoroutine 构建器方法，您可以将异步处理程序转换为可挂起的调用。

这是一个实现读取的示例：

class TcpSocket(private val socket: AsynchronousSocketChannel) {
    suspend fun read(buffer: ByteBuffer): Int {
        return socket.asyncRead(buffer)
    }

    fun close() {
        socket.close()
    }

    private suspend fun AsynchronousSocketChannel.asyncRead(buffer: ByteBuffer): Int {
        return suspendCoroutine { continuation ->
           this.read(buffer, continuation, ReadCompletionHandler)
        }
    }

    object ReadCompletionHandler : CompletionHandler<Int, Continuation<Int>> {
        override fun completed(result: Int, attachment: Continuation<Int>) {
            attachment.resume(result)
        }

        override fun failed(exc: Throwable, attachment: Continuation<Int>) {
            attachment.resumeWithException(exc)
        }
    }
}

您可以选择删除我在这里所做的包装，并像这样在 AsynchronousSocketChannel 上公开一个 asyncRead 方法：

suspend fun AsynchronousSocketChannel.asyncRead(buffer: ByteBuffer): Int {
    return suspendCoroutine { continuation ->
       this.read(buffer, continuation, ReadCompletionHandler)
    }
}

object ReadCompletionHandler : CompletionHandler<Int, Continuation<Int>> {
    override fun completed(result: Int, attachment: Continuation<Int>) {
        attachment.resume(result)
    }

    override fun failed(exc: Throwable, attachment: Continuation<Int>) {
        attachment.resumeWithException(exc)
    }
}

这完全取决于您的品味以及您的设计目标究竟是什么。您应该能够为初始连接实现类似的方法，就像我在此处为读取所做的那样。

Answer 2

// I what to suspend(NOT block) current Java Thread, until connect is success
job.await()

这不是一个现实的期望。从 Java 的角度来看，suspend fun 通过返回一个特殊的常量 COROUTINE_SUSPENDED 来暂停其执行。您需要 Kotlin 编译器将其隐藏起来，并允许您编写可暂停的常规代码。

即使从 Kotlin 的角度来看，您的代码也没有达到非阻塞暂停，因为它使用阻塞调用来连接。将该调用提交给另一个线程并不会使其成为非阻塞。

您的代码所做的完全等同于向 Java 执行器服务提交作业，然后等待其结果。例如，您可以使用CompletableFuture.supplyAsync。