kotlin - 非阻塞 I/O 和 Kotlin 协程有什么关系？

Question

Kotlin协程和非阻塞I/O有什么关系？一个暗示另一个吗？如果我使用阻塞 I/O 会怎样？这对性能有何影响？

Answer 1

协程旨在包含非阻塞(即 CPU 绑定(bind))代码。这就是为什么默认协程调度程序 – Dispatchers.Default – 共有 max(2, num_of_cpus)线程来执行分派(dispatch)的协程。例如，默认情况下，一个高度并发的程序(例如在具有 2 个 CPU 的计算机中运行的 Web 服务器)的计算能力会降低 50%，而线程会阻塞等待 I/O 在协程中完成。

非阻塞 I/O 不是协程的特性。协程只是提供了一个更简单的编程模型，由暂停函数组成，而不是难以阅读CompletableFuture<T> Java 中的延续，以及 structured concurrency以及其他概念。

---

要了解协程和非阻塞 I/O 如何一起工作，这里有一个实际示例:

server.js:接收请求并返回响应的简单 Node.js HTTP 服务器 ~5s之后。

const { createServer } = require("http");

let reqCount = 0;
const server = createServer(async (req, res) => {
    const { method, url } = req;
    const reqNumber = ++reqCount;
    console.log(`${new Date().toISOString()} [${reqNumber}] ${method} ${url}`);
    
    await new Promise((resolve) => setTimeout(resolve, 5000));
    res.end("Hello!\n");
    console.log(`${new Date().toISOString()} [${reqNumber}] done!`);
});

server.listen(8080);
console.log("Server started!");

ma​​in.kt:使用三种实现向 Node.js 服务器发送 128 个 HTTP 请求:

1. withJdkClientBlocking() : 调用 JDK11 java.net.http.HttpClient在 Dispatchers.IO 调度的协程中阻塞 I/O 方法.

import java.net.URI
import java.net.http.HttpClient as JDK11HttpClient
import java.net.http.HttpRequest as JDK11HttpRequest
import java.net.http.HttpResponse as JDK11HttpResponse
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext

fun withJdkClientBlocking() {
    println("Running with JDK11 client using blocking send()")

    val client = JDK11HttpClient.newHttpClient()
    runExample {
        // Sometimes you can't avoid coroutines with blocking I/O methods.
        // These must be always be dispatched by Dispatchers.IO.
        withContext(Dispatchers.IO) {
            // Kotlin compiler warns this is a blocking I/O method.
            val response = client.send(
                JDK11HttpRequest.newBuilder(URI("http://localhost:8080")).build(),
                JDK11HttpResponse.BodyHandlers.ofString()
            )
            // Return status code.
            response.statusCode()
        }
    }
}

2. withJdkClientNonBlocking() : 调用 JDK11 java.net.HttpClient非阻塞 I/O 方法。这些方法返回 CompletableFuture<T>使用 CompletionStage<T>.await() 使用其结果来自 kotlinx-coroutines-jdk8 的互操作性扩展功能.即使 I/O 不阻塞任何线程，异步请求/响应编码/解码在 Java Executor 上运行，因此该示例使用单线程执行器来说明由于非阻塞 I/O，单线程如何处理许多并发请求。

import java.net.URI
import java.net.http.HttpClient as JDK11HttpClient
import java.net.http.HttpRequest as JDK11HttpRequest
import java.net.http.HttpResponse as JDK11HttpResponse
import java.util.concurrent.Executors
import kotlinx.coroutines.future.await

fun withJdkClientNonBlocking() {
    println("Running with JDK11 client using non-blocking sendAsync()")

    val httpExecutor = Executors.newSingleThreadExecutor()
    val client = JDK11HttpClient.newBuilder().executor(httpExecutor).build()
    try {
        runExample {
            // We use `.await()` for interoperability with `CompletableFuture`.
            val response = client.sendAsync(
                JDK11HttpRequest.newBuilder(URI("http://localhost:8080")).build(),
                JDK11HttpResponse.BodyHandlers.ofString()
            ).await()
            // Return status code.
            response.statusCode()
        }
    } finally {
        httpExecutor.shutdown()
    }
}

3. withKtorHttpClient()用途 Ktor ，一个使用 Kotlin 和协程编写的非阻塞 I/O HTTP 客户端。

import io.ktor.client.engine.cio.CIO
import io.ktor.client.HttpClient as KtorClient
import io.ktor.client.request.get
import io.ktor.client.statement.HttpResponse as KtorHttpResponse

fun withKtorHttpClient() {
    println("Running with Ktor client")

    // Non-blocking I/O does not imply unlimited connections to a host.
    // You are still limited by the number of ephemeral ports (an other limits like file descriptors).
    // With no configurable thread limit, you can configure the max number of connections.
    // Note that HTTP/2 allows concurrent requests with a single connection.
    KtorClient(CIO) { engine { maxConnectionsCount = 128 } }.use { client ->
        runExample {
            // KtorClient.get() is a suspend fun, so suspension is implicit here
            val response = client.get<KtorHttpResponse>("http://localhost:8080")
            // Return status code.
            response.status.value
        }
    }
}

综合起来:

import kotlin.system.measureTimeMillis
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.asCoroutineDispatcher
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.runBlocking

fun runExample(block: suspend () -> Int) {
    var successCount = 0
    var failCount = 0

    Executors.newSingleThreadExecutor().asCoroutineDispatcher().use { dispatcher ->
        measureTimeMillis {
            runBlocking(dispatcher) {
                val responses = mutableListOf<Deferred<Int>>()
                repeat(128) { responses += async { block() } }
                responses.awaitAll().forEach {
                    if (it in 200..399) {
                        ++successCount
                    } else {
                        ++failCount
                    }
                }
            }
        }.also {
            println("Successfully sent ${success + fail} requests in ${it}ms: $successCount were successful and $failCount failed.")
        }
    }
}

fun main() {
    withJdkClientBlocking()
    withJdkClientNonBlocking()
    withKtorHttpClient()
}

---

示例运行:

ma​​in.kt(使用 # comments 进行说明)

# There were ~6,454ms of overhead in this execution
Running with JDK11 client using blocking send()
Successfully sent 128 requests in 16454ms: 128 were successful and 0 failed.

# There were ~203ms of overhead in this execution
Running with JDK11 client using non-blocking sendAsync()
Successfully sent 128 requests in 5203ms: 128 were successful and 0 failed.

# There were ~862ms of overhead in this execution
Running with Ktor client
Successfully sent 128 requests in 5862ms: 128 were successful and 0 failed.

server.js(使用 # comments 进行说明)

# These are the requests from JDK11's HttpClient blocking I/O.
# Notice how we only receive 64 requests at a time.
# This is because Dispatchers.IO has a limit of 64 threads by default, so main.kt can't send anymore requests until those are done and the Dispatchers.IO threads are released.
2022-07-24T17:59:29.107Z [1] GET /
(...)
2022-07-24T17:59:29.218Z [64] GET /
2022-07-24T17:59:34.124Z [1] done!
(...)
2022-07-24T17:59:34.219Z [64] done!
2022-07-24T17:59:35.618Z [65] GET /
(...)
2022-07-24T17:59:35.653Z [128] GET /
2022-07-24T17:59:40.624Z [65] done!
(...)
2022-07-24T17:59:40.655Z [128] done!

# These are the requests from JDK11's HttpClient non-blocking I/O.
# Notice how we receive all 128 requests at once.
2022-07-24T17:59:41.163Z [129] GET /
(...)
2022-07-24T17:59:41.257Z [256] GET /
2022-07-24T17:59:46.170Z [129] done!
(...)
2022-07-24T17:59:46.276Z [256] done!

# These are there requests from Ktor's HTTP client non-blocking I/O.
# Notice how we also receive all 128 requests at once.
2022-07-24T17:59:46.869Z [257] GET /
(...)
2022-07-24T17:59:46.918Z [384] GET /
2022-07-24T17:59:51.874Z [257] done!
(...)
2022-07-24T17:59:51.921Z [384] done!