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

Question

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

Answer 1

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

虽然非阻塞 I/O 不是协程的特性。协程只是提供了一个由挂起函数组成的更简单的编程模型，而不是 Java 中难以阅读的 CompletableFuture<T> 延续，以及其他概念中的 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()：在 Dispatchers.IO 调度的协程内调用 JDK11 java.net.http.HttpClient 的阻塞 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>，其结果使用来自kotlinx-coroutines-jdk8 的CompletionStage<T>.await() 互操作性扩展函数使用。即使 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!