有人可以解释为什么 rxjava 在“onNext”中发生异常之后甚至在调用“onError”之后消耗可观察序列中的下一个元素吗？
这是我的模拟：
import java.util.concurrent.{TimeUnit, CountDownLatch}
import rx.lang.scala._
object Tests {
val counter = new CountDownLatch(1)
def buildStream(num: Int) = {
Stream.range(1, num)
.map {s =>
println(s"[${Thread.currentThread().getId}] Taken: $s");
s}
}
val stream = buildStream(10).toSeq
stream.toObservable
.subscribe(
onNext = x => consume(x),
onError = e => println(s"ERROR! $e"),
onCompleted = () => {println("completed"); counter.countDown()}
)
def consume(x: Int) = {
println(s"[${Thread.currentThread().getId}] consuming: $x")
Thread.sleep(100)
if (x == 5) {
throw new Exception(s"[${Thread.currentThread().getId}] consume?! ha!")
}
}
counter.await(10, TimeUnit.SECONDS)
}
其结果如下，您可以看到在引发异常后从序列中取出第 6项：
[39] consuming: 1
[39] Taken: 2
[39] consuming: 2
[39] Taken: 3
[39] consuming: 3
[39] Taken: 4
[39] consuming: 4
[39] Taken: 5
[39] consuming: 5
ERROR! java.lang.Exception: [39] consume?! ha!
[39] Taken: 6
res0: rx.lang.scala.Subscription
据我所知，我从常规序列创建的可观察集合是“冷的”，只有在观察者成功处理当前序列后，才能从源序列中获取下一个项目。我可能会怀疑一些线程问题，但这意味着“冷”不是那种不可能是真的“冷”，而且我可以清楚地看到始终使用相同的线程 ID。
那么为什么从序列中取出第6项？！