我一直在研究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
,并且 this 没有isUnsubscribed()
方法。我查看了What's Different in 2.0并搜索了存储库,但找不到任何此类方法。
如何在 2.0 中实现相同的功能?
编辑以包含如下给出的解决方案(nb using kotlin):
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()