咨询OkHttp3与Ktor实现大量WebSocket连接并限制线程数的方案

一、OkHttp3 适配协程处理大量 WebSocket 连接 #

OkHttp3 完全适用于你的需求，它基于 NIO 实现，本身就不需要为每个连接分配单独线程。结合 Kotlin 协程可以更优雅地处理异步逻辑，避免回调地狱，同时复用线程池。

实现步骤 #

1. 依赖准备：引入 kotlinx-coroutines-okhttp 库，它提供了协程与 OkHttp 的适配工具。
2. 封装 WebSocket 协程流：将 OkHttp 的回调式 WebSocket 转换为协程可感知的流，用 Channel 接收消息，避免每个连接占用线程。
3. 复用线程池：自定义协程调度器，指定固定线程数，让所有连接复用这些线程处理逻辑。

示例代码：

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import okhttp3.*
import java.util.concurrent.Executors

suspend fun main() {
    // 自定义固定线程池的协程调度器
    val dispatcher = Executors.newFixedThreadPool(4).asCoroutineDispatcher()
    val client = OkHttpClient.Builder().build()

    val jobs = mutableListOf<Job>()
    val credentials = listOf("user1", "user2", "...") // 替换为你的用户凭证列表

    credentials.forEach { cred ->
        val job = GlobalScope.launch(dispatcher) {
            val request = Request.Builder()
                .url("wss://stream-testnet.bybit.com/v5/private?max_alive_time=10m")
                .build()

            val messageChannel = Channel<String>()
            val listener = object : WebSocketListener() {
                override fun onOpen(webSocket: WebSocket, response: Response) {
                    // 处理鉴权、订阅逻辑
                    webSocket.send(authMessage(cred))
                    webSocket.send(subscribeMessage())
                }

                override fun onMessage(webSocket: WebSocket, text: String) {
                    runBlocking(dispatcher) {
                        messageChannel.send(text)
                    }
                }

                override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
                    webSocket.close(1000, null)
                    messageChannel.close()
                }

                override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
                    messageChannel.close(t)
                }
            }

            client.newWebSocket(request, listener)

            // 协程内处理消息流
            try {
                for (message in messageChannel) {
                    println("Received message from $cred: $message")
                }
            } catch (e: Exception) {
                println("Connection error for $cred: ${e.message}")
            }
        }
        jobs.add(job)
    }

    jobs.joinAll()
    client.dispatcher.executorService.shutdown()
    dispatcher.close()
}

// 辅助函数：生成鉴权消息
fun authMessage(cred: String): String = "{\"op\":\"auth\",\"args\":[\"$cred\"]}"
// 辅助函数：生成订阅消息
fun subscribeMessage(): String = "{\"op\":\"subscribe\",\"args\":[\"topic\"]}"

二、Ktor CIO 客户端限制 DefaultDispatcher 线程数 #

你遇到的大量 DefaultDispatcher-worker-* 线程问题，是因为 Ktor 的 CIO 引擎内部默认会使用 Dispatchers.Default 处理部分异步操作，比如 IO 事件、消息解析等。要全局限制线程数，需要替换默认的协程调度器，或者配置 CIO 引擎使用自定义线程池。

解决方法 #

1. 全局替换 Dispatchers.Default：在应用启动时，将 Dispatchers.Default 替换为自定义的固定线程池调度器，这样所有依赖 Default 的操作都会复用这些线程。
2. 配置 CIO 引擎使用自定义调度器：显式指定 CIO 引擎的工作调度器，避免它使用默认的 DefaultDispatcher。

修改后的示例代码：

import kotlinx.coroutines.*
import kotlinx.coroutines.asCoroutineDispatcher
import io.ktor.client.*
import io.ktor.client.engine.cio.*
import io.ktor.client.plugins.websocket.*
import io.ktor.http.*
import io.ktor.websocket.*
import java.util.concurrent.Executors

suspend fun main() {
    // 自定义固定线程池的协程调度器
    val customDispatcher = Executors.newFixedThreadPool(4).asCoroutineDispatcher()

    // 全局替换 Dispatchers.Default
    DefaultExecutor.setDispatcher(customDispatcher)

    val jobs = mutableListOf<Job>()
    val credentials = readCsvFile("/creds_4358.csv")
    
    val client = HttpClient(CIO) {
        engine {
            maxConnectionsCount = 10000
            // 指定CIO引擎使用自定义调度器
            dispatcher = customDispatcher
            endpoint {
                maxConnectionsPerRoute = 1000000
                pipelineMaxSize = 20
                keepAliveTime = 5000
                connectTimeout = 5000
                connectAttempts = 5
            }
        }
        install(WebSockets) {
            pingInterval = 2000
            // WebSocket插件也指定自定义调度器
            dispatcher = customDispatcher
        }
    }

    credentials.forEach { cred ->
        val job = GlobalScope.launch(customDispatcher) {
            client.webSocket(
                method = HttpMethod.Get,
                host = "stream-testnet.bybit.com",
                port = 80,
                path = "/v5/private?max_alive_time=10m",
            ) {
                auth(cred)
                subscribe()
                println("I'm working in thread ${Thread.currentThread().name}")
                // 确保消息处理也在自定义调度器上执行
                withContext(customDispatcher) {
                    for (message in incoming) {
                        if (message is Frame.Text) {
                            val text = message.readText()
                            println("Received message from server: $text")
                        }
                    }
                }
                close()
            }
        }
        jobs.add(job)
    }

    jobs.joinAll()
    client.close()
    customDispatcher.close()
    println("Connection closed. Goodbye!")
}

关键说明 #

• DefaultExecutor.setDispatcher(customDispatcher)：替换全局的 Dispatchers.Default，确保所有依赖默认调度器的操作都使用自定义线程池。
• CIO 引擎的 dispatcher 配置：显式指定引擎内部处理 IO 事件的调度器。
• WebSocket 插件的 dispatcher 配置：指定 WebSocket 消息处理的调度器。
• withContext(customDispatcher)：确保消息循环在自定义线程上执行，避免切换到默认调度器。

内容的提问来源于stack exchange，提问作者advortsov