Kotlin非Android环境下任务调度器实现方案正确性验证

这个初始实现不符合需求且存在多处潜在问题，具体问题和修正方案如下：

一、核心问题点 #

1. SupervisorJob复用导致重启失败 #

类级别的supervisorJob一旦被stop()方法取消后，无法再次复用。后续调用start()时，基于已取消的Job启动新协程，会直接导致新任务被取消，无法正常执行。

2. 任务上下文绑定错误 #

在supervisorScope中启动任务时，错误地绑定了类级别的supervisorJob，导致supervisorScope的作用失效——原本supervisorScope的子协程失败不会互相影响，但绑定外部Job后，这个隔离特性丢失。

3. stop()方法逻辑不完整 #

stop()仅设置isRunning=false并取消子Job，但未取消start()方法所在的调度协程，可能导致调度循环继续执行；同时jobList的清理未考虑并发场景，存在线程安全风险。

4. 任务超时异常的处理冗余 #

withTimeout抛出的TimeoutCancellationException属于协程正常取消信号，不需要在start()的catch块中单独处理，当前的捕获逻辑会干扰协程的正常取消流程。

二、修正后的代码实现 #

import kotlinx.coroutines.*
import java.time.Duration
import java.time.LocalDateTime
import java.time.LocalTime
import java.util.concurrent.CopyOnWriteArrayList

class JobScheduler(private val repository: Repository) {

    private var isRunning = false
    private val jobList = CopyOnWriteArrayList<Job>()
    private var schedulerJob: Job? = null

    suspend fun start(startHour: Int, isImmediately: Boolean, numOfJobs: Int) {
        if (isRunning) return // 避免重复启动
        isRunning = true

        val coroutineExceptionHandler = CoroutineExceptionHandler { _, throwable ->
            println("任务执行异常: $throwable")
        }

        schedulerJob = supervisorScope {
            launch(Dispatchers.Default) {
                var waitTime = if (isImmediately) Duration.ZERO else startHour.amountTimeFromNow()

                while (isRunning) {
                    try {
                        // 等待到启动时间
                        delay(waitTime.toMillis())

                        // 取消所有之前的任务
                        jobList.forEach { it.cancel() }
                        jobList.clear()

                        // 计算任务最大执行时长（当前时间到下一个startHour的间隔）
                        val maxJobDuration = startHour.amountTimeFromNow().toMillis()

                        // 启动并行任务
                        val newJobs = (1..numOfJobs).map {
                            launch(coroutineExceptionHandler) {
                                withTimeout(maxJobDuration) {
                                    while (isActive) {
                                        val success = repository.call()
                                        if (success) break
                                        delay(Duration.ofSeconds(5).toMillis())
                                    }
                                }
                            }
                        }
                        jobList.addAll(newJobs)

                        // 更新下次启动的等待时间
                        waitTime = startHour.amountTimeFromNow()

                    } catch (e: Exception) {
                        if (e !is CancellationException) {
                            println("调度器异常: $e")
                        }
                    }
                }
            }
        }
    }

    fun stop() {
        isRunning = false
        schedulerJob?.cancel()
        jobList.forEach { it.cancel() }
        jobList.clear()
    }

    private fun Int.amountTimeFromNow(): Duration {
        val now = LocalDateTime.now()
        val targetTime = LocalDateTime.of(now.toLocalDate(), LocalTime.of(this, 0))
        val nextTarget = if (now.isAfter(targetTime) || now.isEqual(targetTime)) {
            targetTime.plusDays(1)
        } else {
            targetTime
        }
        return Duration.between(now, nextTarget)
    }
}

// 假设Repository接口定义
interface Repository {
    suspend fun call(): Boolean
}

三、修正说明 #

1. Job管理优化：

   • 新增schedulerJob保存调度协程的引用，stop()时直接取消整个调度流程，确保彻底终止。
   • 使用CopyOnWriteArrayList存储任务Job，避免并发修改导致的异常。

2. 协程上下文修正：

   • 在supervisorScope内部启动任务，直接使用coroutineExceptionHandler，保留supervisorScope的子协程隔离特性——单个任务失败/超时不会影响其他任务。

3. 启动逻辑优化：

   • 增加重复启动判断，避免多次调用start()导致的重复调度。
   • 将while(true)改为while(isActive)，利用协程的活跃状态判断，更符合协程的取消语义。

4. 异常处理简化：

   • 仅处理非取消类异常，避免干扰协程的正常取消流程。

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