Kotlin协程Job取消失效：为何自定义实现可正常工作？

为什么你的初始代码无法正确取消协程？ #

你的CancellationTest类存在两个核心问题，导致旧协程没被真正取消，进而出现重复发送数据的情况：

1. cancel()是异步非阻塞操作
   调用job?.cancel()仅向协程发送取消信号，不会等待协程实际停止执行。如果在调用cancel()后立刻启动新协程，旧协程可能还在执行doWork()里的挂起函数（比如尚未走到支持取消的挂起点），会继续完成数据发送操作，最终导致重复提交。

2. 并发场景下的竞态条件
   你在for循环中快速调用run()，多次调用可能同时操作job变量：比如前一次的cancel()还未生效，后一次的run()已经将job替换为新的协程对象，旧协程的引用丢失，后续无法再对它执行取消操作，它会持续执行直到结束。

为什么WorkingCancellation能正常工作？ #

这个类通过以下关键点解决了上述问题：

1. 用cancelAndJoin()等待取消完成
   activeTask.get()?.cancelAndJoin()会阻塞当前协程，直到旧任务真正被取消并结束，确保旧任务完全停止后才启动新任务，从根源避免了新旧任务并行执行。

2. 原子类保证线程安全
   借助AtomicReference<Deferred<T>?>管理活跃任务，避免了并发场景下多个线程同时修改任务引用的竞态条件，确保同一时间只有一个任务处于活跃状态。

3. CAS循环处理并发竞争
   while循环中的compareAndSet(null, newTask)是核心逻辑：

   • 如果CAS失败，说明此时有其他任务正在启动，就取消该任务并重新尝试
   • 如果CAS成功，才启动当前新任务并等待它完成
     这种机制确保即使在高并发调用下，也不会有多个任务同时运行。

4. 任务完成后主动清理引用
   通过newTask.invokeOnCompletion { activeTask.compareAndSet(newTask, null) }，在任务完成后清理活跃任务的引用，既避免内存泄漏，也确保下一次调用能正确获取最新的任务状态。

修复初始代码的简化方案 #

如果不想依赖WorkingCancellation，可以修改run()方法解决核心问题：

class CancellationTest {
    private var job: Job? = null
    private val scope = MainScope()
    private val jobLock = Any()

    suspend fun run() {
        synchronized(jobLock) {
            job?.cancelAndJoin() // 等待旧任务彻底取消
            job = scope.launch { doWork() }
        }
    }

    private suspend fun doWork() {
        // 支持取消的挂起API调用逻辑
    }
}

这里将run()改为挂起函数（因为cancelAndJoin()是挂起函数），同时用synchronized保证并发调用时的线程安全，避免竞态条件。

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