ExecutorCoroutineDispatcher正确实现：解决死锁与栈溢出问题

原始代码 #

class Dispatch : ExecutorCoroutineDispatcher() {
    private val services = Executors.newCachedThreadPool()

    override val executor: Executor
        get() = services

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        println("dispatch ")
        if(this.isDispatchNeeded(context)){
            executor.execute(block)
        }else{
            Dispatchers.Unconfined.dispatch(context , block) 
        }
    }

    override fun isDispatchNeeded(context: CoroutineContext): Boolean {
        println("isDispatchedNeeded ")
        // Implement your custom logic here to determine if dispatch is needed
        return false // assuming yield() call in loop so return false 
    }

    override fun close() {
        services.shutdown()
    }
}

fun main() {
    runBlocking {
        launch(Dispatch()) {
            for (i in 1..3) {
                println("start")
                yield()// dispatch() call in loop may cause StackOverflowError
                println("end")
            }
        }

        launch(Dispatch()) {
            println("some suspend function")
            work() // some suspend work 
        }
    }
}

suspend fun work() {
    delay(1000) // Simulating some suspend work
}

问题说明 #

我参考了dispatch()的官方文档，但不确定如何正确实现该方法。文档指出：

此方法必须保证给定的block最终被调用，否则系统可能陷入死锁无法退出。
此方法不得直接调用block，否则在dispatch被重复调用时（例如循环中调用yield()）可能导致StackOverflowError。如需就地执行block，需从isDispatchNeeded返回false，并将调度委托给Dispatchers.Unconfined.dispatch，协程机制会确保就地执行并形成事件循环以避免无限递归。

我的代码中，为了避免循环内yield()导致的栈溢出，让isDispatchNeeded固定返回false并使用Unconfined调度器，但不清楚如何在需要立即执行block时避免死锁。需要修正代码，确保能正确处理协程调度（兼顾dispatch()被循环调用的场景），并完善dispatch()、isDispatchNeeded()的错误处理，解决死锁、StackOverflowError等问题。

修正后的实现 #

问题根源 #

1. isDispatchNeeded逻辑僵化：固定返回false导致所有任务都走Unconfined调度，完全浪费了自定义线程池的作用，还可能引发线程安全问题。
2. 缺乏异常防护：线程池执行任务时未处理异常，会导致任务静默失败；close()方法未处理线程池关闭超时或中断的情况。
3. 死锁隐患：如果isDispatchNeeded在需要调度的场景错误返回false，会导致任务无法被线程池执行，阻塞整个协程流程。

修正代码 #

import kotlinx.coroutines.*
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.concurrent.RejectedExecutionException

class CustomDispatcher : ExecutorCoroutineDispatcher() {
    // 用固定线程池替代CachedThreadPool，避免无限制创建线程
    private val executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors())

    override val executor: Executor
        get() = executorService

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        try {
            if (isDispatchNeeded(context)) {
                // 提交任务到线程池，包裹异常处理防止线程池线程崩溃
                executor.execute {
                    runCatching {
                        block.run()
                    }.onFailure { e ->
                        println("任务执行异常: ${e.message}")
                        e.printStackTrace()
                    }
                }
            } else {
                // 委托Unconfined处理就地执行，避免栈溢出
                Dispatchers.Unconfined.dispatch(context, block)
            }
        } catch (e: RejectedExecutionException) {
            // 线程池关闭后任务被拒绝，降级到Unconfined确保任务执行
            println("线程池已关闭，任务将就地执行: ${e.message}")
            Dispatchers.Unconfined.dispatch(context, block)
        } catch (e: Throwable) {
            println("调度流程异常: ${e.message}")
            e.printStackTrace()
        }
    }

    override fun isDispatchNeeded(context: CoroutineContext): Boolean {
        return runCatching {
            // 自定义调度判断：
            // 1. 当前线程不是自定义线程池的线程 → 需要调度
            // 2. 当前是Unconfined线程 → 需要调度到自定义池
            val currentThread = Thread.currentThread()
            !currentThread.name.startsWith("pool-") || Dispatchers.Unconfined.isDispatchNeeded(context)
        }.getOrDefault(true) // 异常时默认返回true，确保任务被调度，避免死锁
    }

    override fun close() {
        runCatching {
            executorService.shutdown()
            // 等待线程池优雅关闭，超时则强制终止
            if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                executorService.shutdownNow()
                println("线程池强制关闭")
            } else {
                println("线程池正常关闭")
            }
        }.onFailure { e ->
            println("线程池关闭异常: ${e.message}")
            e.printStackTrace()
            executorService.shutdownNow()
        }
    }
}

fun main() = runBlocking {
    val dispatcher = CustomDispatcher()
    try {
        launch(dispatcher) {
            for (i in 1..3) {
                println("任务1 - 第$i轮 start | 线程: ${Thread.currentThread().name}")
                yield()
                println("任务1 - 第$i轮 end | 线程: ${Thread.currentThread().name}")
            }
        }

        launch(dispatcher) {
            println("任务2 - 执行挂起函数 | 线程: ${Thread.currentThread().name}")
            work()
            println("任务2 - 挂起函数完成 | 线程: ${Thread.currentThread().name}")
        }
    } finally {
        // 确保调度器资源被释放
        dispatcher.close()
    }
}

suspend fun work() {
    delay(1000) // 模拟挂起操作
}

核心优化说明 #

• 智能调度判断：isDispatchNeeded会检查当前线程是否属于自定义线程池，避免重复调度；异常时默认返回true，确保任务被执行，彻底消除死锁风险。
• 全链路异常处理：
  • 任务执行时用runCatching捕获所有异常，防止线程池线程因未处理异常终止。
  • 调度流程中处理线程池拒绝任务的情况，降级到Unconfined保证任务不丢失。
  • 线程池关闭时处理超时和中断，确保资源正常释放。
• 线程池优化：改用newFixedThreadPool并基于CPU核心数设置线程数，避免CachedThreadPool无限创建线程的风险，更高效利用系统资源。
• 资源安全保障：main函数中用finally块确保调度器被关闭，避免线程泄漏。

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