如何限制协程并行数量以适配固定大小的数据库连接池？

你的问题核心在于：即使单线程Dispatcher下，协程的并发切换会导致多个协程同时占用数据库连接，超过连接池上限。下面是几种靠谱的解决方案：

1. 用信号量（Semaphore）精准控制并发数 #

直接用协程的Semaphore来限制同时执行DB操作的协程数量，许可数设为连接池的大小，确保同一时间只有指定数量的协程能获取连接。

示例代码：

// 初始化信号量，许可数等于HikariCP连接池大小
val dbSemaphore = Semaphore(hikariPoolSize)

entities.forEach { entity ->
    launch {
        // 自动申请和释放许可
        dbSemaphore.withPermit {
            doingSomethingWithDB(entity)
        }
    }
}

suspend fun doingSomethingWithDB(entity) {
    val conn = getDBConnectionFromPool()
    try {
        // 执行数据库操作逻辑
    } finally {
        // 确保连接一定会归还，防止泄漏
        returnDBConnectionToPool(conn)
    }
}

2. 用Dispatcher的limitedParallelism限制并行度 #

给DB操作单独分配一个带有固定并行度的Dispatcher，并行度和连接池大小一致，所有DB相关协程都跑在这个Dispatcher上，从调度层面限制并发数。

示例代码：

// 创建并行度等于连接池大小的Dispatcher
val dbDispatcher = Dispatchers.IO.limitedParallelism(hikariPoolSize)

entities.forEach { entity ->
    // 指定用dbDispatcher执行协程
    launch(dbDispatcher) {
        doingSomethingWithDB(entity)
    }
}

suspend fun doingSomethingWithDB(entity) {
    val conn = getDBConnectionFromPool()
    try {
        // 执行数据库操作逻辑
    } finally {
        returnDBConnectionToPool(conn)
    }
}

3. 用Flow批量处理（适合大量数据场景） #

如果要处理的entities数量很大，用Flow的并发操作符来控制并行度，代码更简洁且适合流式处理：

entities.asFlow()
    // 指定并发处理的数量等于连接池大小
    .flatMapMerge(concurrency = hikariPoolSize) { entity ->
        flow {
            doingSomethingWithDB(entity)
        }
    }
    .collect()

关键注意点 #

• 必须用try-finally包裹连接的使用，确保协程被取消时连接也能正确归还到池里，避免连接泄漏。
• 信号量方式更灵活，能针对特定代码块做限制；Dispatcher方式则是全局限制该调度器上的所有协程，按需选择。

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