kotlin - 与 onException Resume Next 混淆，将可观察对象作为 lambda 表达式传递

Question

io.reactivex.rxjava2:rxjava:2.1.13
kotlin_version = '1.2.30'

我有以下 Observable，我试图抛出一个异常来测试 OnError 中异常的捕获。但是，当我将以下内容传递给 onExceptionResumeNext(Observable.just(10)) 时，我得到以下输出:

1
2
10
onComplete

fun main(args: Array<String>) {
    Observable.fromArray(1, 2, 0, 4, 5, 6)
            .doOnNext {
                if (it == 0) {
                    throw RuntimeException("Exception on 0")
                }
            }
            .onExceptionResumeNext(Observable.just(10))
            .subscribe(
                    {
                        println(it)
                    },
                    {
                        println("onError ${it.message}")
                    },
                    {
                        println("onComplete")
                    } )
}

但是，如果我将 lambda 表达式传递给该方法，我会得到以下输出:

1
2

 Observable.fromArray(1, 2, 0, 4, 5, 6)
            .doOnNext {
                if (it == 0) {
                    throw RuntimeException("Exception on 0")
                }
            }
            .onExceptionResumeNext { Observable.just(10) }
            .subscribe(
                    {
                        println(it)
                    },
                    {
                        println("onError ${it.message}")
                    },
                    {
                        println("onComplete")
                    } )
}

只是想知道这样做 onExceptionResumeNext(Observable.just(10)) 和这样做有什么区别 onExceptionResumeNext { Observable.just(10) }

提前致谢

Answer 1

在第一种情况下，您将 observable 传递给 onExceptionResumeNext

如果原始 observable 出现错误，您将重新订阅传递给 onExceptionResumeNext

的 observable

这就是为什么您从原始来源获得“1”和“2”，然后从 Observable.just(10)

获得“10”和“onComplete”

在第二种情况下，您使用的是 lambda 语法，因此您实际上是在这样做:

.onExceptionResumeNext(object : ObservableSource {
    void subscribe(@NonNull Observer<? super T> observer) {
        Observable.just(10)
    }
})

如您所见，传递给您的 observer 没有被订阅。因此，在初始源的“1”和“2”之后没有任何内容

你应该做的是

.onExceptionResumeNext { observer ->
    Observable.just(10).subscribe(observer)
}