RxJava 2相当于isUnsubscribed

Question

我一直在努力通过Reactive Programming with RxJava这本书的例子，该书的目标不是版本1而是2.对无限流的介绍有以下示例（并指出有更好的方法来处理并发）：

Observable<BigInteger> naturalNumbers = Observable.create(subscriber -> { 
    Runnabler =() -> { 
     BigInteger i = ZERO; 
     while (!subscriber.isUnsubscribed()) { 
      subscriber.onNext(i); 
      i = i.add(ONE); 
     } 
    }; 
    new Thread(r).start(); 
}); 

... 

Subscription subscription = naturalNumbers.subscribe(x -> log(x)); 
/* after some time... */ 
subscription.unsubscribe(); 

然而，在RxJava 2，传递给create()方法lambda表达式是ObservableEmitter类型的，并且这不具有isUnsubscribed()方法。我看了一下What's Different in 2.0，同时也搜索了版本库，但是找不到任何这样的方法。

如何在2.0中实现相同的功能？

为如下（注意，使用科特林）编辑，包括解决方案：

val naturalNumbers = Observable.create<BigInteger> { emitter -> 
    Thread({ 
     var int: BigInteger = BigInteger.ZERO 
     while (!emitter.isDisposed) { 
      emitter.onNext(int) 
      int = int.add(BigInteger.ONE) 
     } 
    }).start() 
} 

val first = naturalNumbers.subscribe { log("First: $it") } 
val second = naturalNumbers.subscribe { log("Second: $it") } 

Thread.sleep(5) 
first.dispose() 
Thread.sleep(5) 
second.dispose() 

Answer 1

后您订阅观测到的，则返回Disposable。您可以将它保存到您的本地变量并检查disposable.isDisposed()以查看它是否仍订阅。