正确取消 kotlin 协程作业

Question

我正在为如何正确取消协程工作而摸不着头脑。测试用例很简单我有一个包含两种方法的类：

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

   fun run() {
      job?.cancel()
      job = scope.launch { doWork() }
   }

   fun doWork() {
      // gets data from some source and send it to BE
   }
}

方法doWork有一个 api 调用，并且suspending遵循cancellation.

在上面的示例中，在对成功发送到后端的对象进行计数后，我可以看到许多重复项，这意味着之前cancel没有真正cancel调用过。

但是，如果我使用在互联网上找到的片段

internal class WorkingCancellation<T> {
    private val activeTask = AtomicReference<Deferred<T>?>(null)
    suspend fun cancelPreviousThenRun(block: suspend () -> T): T {
        activeTask.get()?.cancelAndJoin()

        return coroutineScope {
            val newTask = async(start = CoroutineStart.LAZY) {
                block()
            }

            newTask.invokeOnCompletion {
                activeTask.compareAndSet(newTask, null)
            }

            val result: T

            while (true) {
                if (!activeTask.compareAndSet(null, newTask)) {
                    activeTask.get()?.cancelAndJoin()
                    yield()
                } else {
                    result = newTask.await()
                    break
                }
            }

            result
        }
    }
}

它工作正常，对象不会重复并正确发送到 BE。最后一件事是我run在 a 中调用方法for loop- 但无论如何我不确定我理解为什么job?.cancel不能正确完成其工作并且WorkingCancellation实际上正在工作

Answer 1

简短的回答：只有当您调用挂起库函数时，取消才可以开箱即用。非挂起代码需要手动检查才能使其可取消。

Kotlin 协程中的取消是合作性的，需要取消作业来检查取消情况并终止其正在执行的任何工作。如果作业不检查取消，它可以很高兴地继续运行，永远不会发现它已被取消。

当您调用内置挂起函数时，协程会自动检查是否取消。如果您查看常用挂起函数（如await()和）的文档yield()，您会发现它们总是说“此挂起函数是可取消的”。

您doWork不是一个suspend函数，因此它无法调用任何其他挂起函数，因此永远不会执行这些自动检查之一以进行取消。如果您确实想取消它，则需要让它定期检查作业是否仍然处于活动状态，或者更改其实现以使用挂起功能。ensureActive您可以通过拨打 来手动检查取消情况Job。

Answer 2

除了 Sam 的回答之外，请考虑这个模拟连续事务的示例，例如服务器的位置更新。

var pingInterval = System.currentTimeMillis()
job = launch {
     while (true) {
         if (System.currentTimeMillis() > pingInterval) {
             Log.e("LocationJob", "Executing location updates... ")
             pingInterval += 1000L
         }
     }
}

它会不断地用位置更新来“ping”服务器，或者像任何其他常见的用例一样，说这会不断地从中获取一些东西。

然后我这里有一个函数，由取消此操作的按钮调用job。

fun cancel() {
    job.cancel()
    Log.e("LocationJob", "Location updates done.")
}

当调用此函数时，会job被取消，但是操作会继续进行，因为没有任何东西可以确保协程作用域停止工作，上面的所有操作都会打印

 Ping server my location...
 Ping server my location...
 Ping server my location...
 Ping server my location...
 Location updates done.
 Ping server my location...
 Ping server my location...

现在如果我们插入ensureActive()无限循环

while (true) {
     ensureActive()
     if (System.currentTimeMillis() > pingInterval) {
          Log.e("LocationJob", "Ping server my location... ")
           pingInterval += 1000L
     }
}

取消job将保证操作将停止。我测试了使用delay，但它保证了取消调用cancellation时的总数。job放置ensureActive()，并在 2 秒后取消，打印

 Ping server my location...
 Ping server my location...
 Location updates done.