Coroutines和Rxjava异步编程对比之二

原文地址：https://medium.com/@manuelvicnt/coroutines-and-rxjava-an-asynchronicity-comparison-part-2-cancelling-execution-199485cdf068

在Android开发中，取消RxJava或Coroutines的执行时很重要的，因为在View即将被销毁的时候，取消网络请求或者对象的创建是必须要进行的操作。

Rxjava

我们使用interval操作符创建一个定时器

Observable.interval(1, TimeUnit.SECONDS)

当你订阅这个可观察的对象时，定时器将启动，并且它将在订阅后每秒发送一个事件给订阅者。

如何取消这个定时器呢

当你订阅了这个定时器，会返回一个Disposable对象。

val disposable: Disposable = 
    Observable.interval(1, TimeUnit.SECONDS).subscribe()

然后你再Disposable对象上调用dispose()方法就会终止定时器的运行。

注意事项

如果你不使用任何操作符，手动创建Observable对象，就不需要处理取消操作。
这个Observable没有准备好被取消，我们在执行取消操作之前需要先检查发射器是否依然在订阅。

Observable.create<Int> { emitter ->
   for (i in 1..5) {
       if (!emitter.isDisposed) {
           emitter.onNext(i)
       } else {
           break
       }
   }   
   emitter.onComplete()
}

Coroutines

Coroutines的Job提供了一个cancel的方法。
例如下面的示例

val job = launch(CommonPool) {
    // my suspending block
}

job.cancel()

某些协程生成器（例如launch和async）采用一个名为parent的命名参数，您可以使用该参数为将要创建的协程设置Job。

val parentJob = Job()

async(CommonPool, parent = parentJob) {
   // my suspending block
}
parentJob.cancel()

这种方式的好处就是你可以在多个协程中共享Job，如果你调用parentJob.cancel()的话，那么以该job作为参数的所有的协程都将被取消。
这种方式有点类似于Rxjava的CompositeDisposable可以同时取消多个订阅。

val parentJob = Job()

val deferred1 = async(CommonPool, parent = parentJob) {
    // my suspending block
}

val deferred2 = async(CommonPool, parent = parentJob) {
    // my suspending block
}

parentJob.cancel()

另一种方式是将Coroutines上下文组合起来，使用加号进行操作。

val parentJob = Job()

launch(parentJob + CommonPool) {
   // my suspending block
}
parentJob.cancel()

在这种情况下，线程策略将根据CommonPool和parentJob中的Job值进行定义。 read this part of the Kotlin Coroutines documentation.

注意事项

像Rxjava一样，你也需要考虑Coroutines的取消。

val job = launch(CommonPool) {
    for (i in 1..5) {
        heavyComputation()
    }
}
job.cancel()

执行这段代码，他会重复执行五次复杂的运算在没有被取消的情况下。

如何改进？
就像在Rxjava中我们需要检查是否订阅，Coroutines中我们需要检查Coroutines是否活跃。

val job = launch(CommonPool) {
    for (i in 1..5) {
        if (!isActive) { break }
        heavyComputation()
    }
}
job.cancel()

isActive是一个内部变量，可以在Coroutines中进行访问。
标准协程库中的一些暂停的函数，可以为我们执行取消操作。

val job = launch(CommonPool) {
    doSomething()
    delay(300) // It’s going to cancel at this point
    doSomething()
}
job.cancel()

Delay是一个暂停的函数，可以为我们处理取消操作，如果你使用Thread.sleep()来代替Delay，那么将会使线程阻塞并不会终止Coroutines。

val job = launch(CommonPool) {
    doSomething()
    Thread.sleep(300) // It’s NOT going to cancel execution
    doSomething()
}
job.cancel()

Thread.sleep不会取消我们的执行。这甚至不是暂停功能！即使我们调用job.cancel()，该协程也不会被取消。
Thread.sleep不是你应该在这个实例中使用的东西。如果你真的需要，取消该协程的一种方法是在线程块之前和之后检查它是否处于活动状态。

val job = launch(CommonPool) {
    doSomething()
    if (!isActive) return
    Thread.sleep(300) // It’s NOT going to cancel execution
    if (!isActive) return
    doSomething()
}
job.cancel()