Spring Boot应用如何用协程/CompletableFuture提升吞吐量？

原有阻塞代码 #

原有代码串行执行两个重型数据库查询和一个外部API调用，性能瓶颈明显：

override fun getTotalIncomeByMonth(retailerMsisdn: String, month: Int, year: Int): SummaryDto {

    val incomes1: TransactionSummaryDao? =
        service1.func(
            retailerMsisdn, month, year
        ) // 重型数据库查询（阻塞）

    val income2: Double = service2.func(
        retailerMsisdn, month, year
    ) // 另一个重型数据库查询（阻塞）

    val totalSimSale: Double = externalApiService.apiCall(
        retailerMsisdn, month, year
    )  // OkHttp外部API调用（阻塞）

    return SummaryDto().apply {
        this.incomeX = incomes1
        this.incomeY = income2
        this.simSale = totalSimSale
    }
}

线程池方案代码 #

尝试用ThreadPoolTaskExecutor实现并发，但存在明显缺陷：

override fun getTotalIncomeByMonth(retailerMsisdn: String, month: Int, year: Int): SummaryDto {

    val executor = ThreadPoolTaskExecutor()
    executor.corePoolSize = 2
    executor.maxPoolSize = 2
    executor.setQueueCapacity(3)
    executor.setTaskDecorator(MdcTaskDecorator())
    executor.setThreadNamePrefix(Constant.NAME_ASYNC_THREAD_PREFIX)
    executor.setWaitForTasksToCompleteOnShutdown(true)
    executor.initialize()

    val incomes1: CompletableFuture<Dao> = CompletableFuture.supplyAsync(
        {
            service1.func(
                retailerMsisdn, month, year
            ) // 重型数据库查询（IO阻塞）
        }, executor )

    val income2: CompletableFuture<Double>  = CompletableFuture.supplyAsync(
        {
            service2.func(
               retailerMsisdn, month, year
            ) // 另一个重型数据库查询（IO阻塞）
        }, executor) 

    val totalSale = externalApiService.apiCall(
        retailerMsisdn, month, year
    )  // OkHttp API调用（未加入并发）

    return SummaryDto().apply {
        this.incomeX = incomes1.join()
        this.incomeY = income2.join()
        this.simSale = totalSale // 原文笔误修正
    }
}

Kotlin协程方案代码 #

尝试用协程实现并发，但未优化阻塞IO的调度：

override fun getTotalIncomeByMonth(retailerMsisdn: String, month: Int, year: Int): SummaryDto {
    return runBlocking {
        val income1: Dao? = async {
            service1.func(retailerMsisdn, month, year)
            // 重型数据库查询（阻塞）
        }

        val income2: Double = async {
            service2.func(retailerMsisdn, month, year)
            // 另一个重型数据库查询（阻塞）
        }

        val simSale: Double = async {
            externalApiService.apiCall(retailerMsisdn, month, year)
            // OkHttp API调用（阻塞）
        }

        SummaryDto().apply {
            this.incomeX = income1.await()
            this.incomeY = income2.await()
            this.simSale = simSale.await()
        }
    }
}

核心问题 #

1. 上述线程池方案是否可行？
2. 应优先采用Kotlin协程+挂起函数方案？
3. 能否在不使用额外调度线程的情况下处理这3个IO操作？
4. 如何兼顾百万级用户、每秒千级请求的场景，避免线程创建开销？

方案分析与优化建议 #

1. 线程池方案的致命缺陷 #

当前线程池方案完全不可行，核心问题如下：

• 每次请求创建新线程池：违背线程池复用的核心设计，每秒千级请求会创建上千个线程池，导致线程数爆炸，引发OOM或严重的上下文切换开销。正确做法是将线程池定义为Spring单例Bean，全局复用。
• 并发不完整：外部API调用仍为串行执行，未加入线程池并发，无法充分利用多核资源。
• 参数设置不合理：core/maxPoolSize=2、队列容量3，对于每秒千级请求来说，队列会迅速堆满，触发任务拒绝，直接导致请求失败。IO密集型场景下，线程池线程数建议设置为CPU核心数*2或更高，队列容量需根据请求量调整（或使用无界队列，但需监控内存避免OOM）。

2. Kotlin协程方案的优化方向 #

当前协程方案未充分发挥协程优势，需针对阻塞IO做优化：

• 阻塞IO必须切换到IO调度器：直接在async中调用阻塞方法会占用协程调度线程（默认Dispatchers.Default），导致线程阻塞，无法高效复用。需用withContext(Dispatchers.IO)包装阻塞操作，该调度器专为IO密集型任务设计，会动态复用线程，减少线程创建开销。
• 推荐将服务方法改为挂起函数：把阻塞的数据库查询、API调用封装为挂起函数，上层代码更简洁，且调度逻辑内聚。

优化后的协程代码示例：

// 先将服务方法改为挂起函数
suspend fun Service1.func(retailerMsisdn: String, month: Int, year: Int): TransactionSummaryDao? =
    withContext(Dispatchers.IO) {
        // 原有阻塞数据库查询逻辑
    }

suspend fun Service2.func(retailerMsisdn: String, month: Int, year: Int): Double =
    withContext(Dispatchers.IO) {
        // 原有阻塞数据库查询逻辑
    }

suspend fun ExternalApiService.apiCall(retailerMsisdn: String, month: Int, year: Int): Double =
    withContext(Dispatchers.IO) {
        // 原有OkHttp阻塞调用逻辑
    }

// 服务层代码
override fun getTotalIncomeByMonth(retailerMsisdn: String, month: Int, year: Int): SummaryDto {
    return runBlocking {
        val income1 = async { service1.func(retailerMsisdn, month, year) }
        val income2 = async { service2.func(retailerMsisdn, month, year) }
        val simSale = async { externalApiService.apiCall(retailerMsisdn, month, year) }

        SummaryDto().apply {
            this.incomeX = income1.await()
            this.incomeY = income2.await()
            this.simSale = simSale.await()
        }
    }
}

3. 能否不使用额外调度线程处理IO？ #

不可能：阻塞IO操作本质上会占用操作系统线程，因为IO等待时线程会被挂起。但协程的Dispatchers.IO可以最大化线程复用：当一个协程等待IO时，调度器会将线程分配给其他活跃协程，实际占用的线程数远低于请求数（比如每秒千级请求，仅需几十条线程即可处理），比传统线程池的资源利用率高得多。

4. 高并发场景的选型结论 #

针对百万级用户、每秒千级请求的场景，优先选择Kotlin协程+挂起函数方案，原因如下：

• 更低的上下文切换开销：协程切换是用户态操作，开销远低于线程的内核态切换，适合高并发场景。
• 自动线程管理：Dispatchers.IO自动根据IO负载调整线程数，无需手动调优线程池参数，避免参数不合理导致的问题。
• 代码可读性更高：协程的异步逻辑用同步写法实现，比CompletableFuture的链式调用更直观，维护成本更低。
• 资源利用率更高：用最少的线程处理最多的IO请求，避免线程数膨胀导致的资源耗尽。

同时需配合基础优化：

• 优化数据库查询索引，减少重型查询的执行时间。
• 配置OkHttp连接池，复用HTTP连接，降低握手开销。
• 监控系统指标（线程数、协程调度情况、数据库连接池状态），及时排查瓶颈。

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