RxJava 默认错误处理程序

Question

抱歉，是否可以在 RxJava 中设置默认错误处理程序？

例如，我在Utils.kt文件中的一些代码：

fun BaseFragment.callGallery(view: View){
    view.clicks().bindToLifecycle(this).subscribe {
        RxPaparazzo.takeImage(this)
        .usingGallery()
        .subscribe { response ->
            throw RuntimeException("Where is this exception from?")
        }
    }
}

但在 stacktrace 中没有任何关于 Utils.kt 或我的任何文件的提示。

我知道我可以在每个订阅者中设置 onError。代码如下：

.subscribe ({ response ->
    ....
}, { it.printStackTrace() })

但我更愿意为所有订阅者设置一个默认的 onError。我该怎么做？

堆栈跟踪：

05-07 12:11:48.246 10966-10966/ru.egslava.rxfluxtest E/AndroidRuntime: FATAL EXCEPTION: main
 Process: ru.egslava.rxfluxtest, PID: 10966
 java.lang.RuntimeException: Unable to destroy activity {ru.egslava.rxfluxtest/rx_activity_result.HolderActivity}: rx.exceptions.OnErrorNotImplementedException: Where is this exception from?
     at android.app.ActivityThread.performDestroyActivity(ActivityThread.java:3831)
     at android.app.ActivityThread.handleDestroyActivity(ActivityThread.java:3849)
     at android.app.ActivityThread.-wrap5(ActivityThread.java)
     at android.app.ActivityThread$H.handleMessage(ActivityThread.java:1398)
     at android.os.Handler.dispatchMessage(Handler.java:102)
     at android.os.Looper.loop(Looper.java:148)
     at android.app.ActivityThread.main(ActivityThread.java:5417)
     at java.lang.reflect.Method.invoke(Native Method)
     at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:726)
     at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:616)
  Caused by: rx.exceptions.OnErrorNotImplementedException: Where is this exception from?
     at rx.internal.util.InternalObservableUtils$ErrorNotImplementedAction.call(InternalObservableUtils.java:374)
     at rx.internal.util.InternalObservableUtils$ErrorNotImplementedAction.call(InternalObservableUtils.java:371)
     at rx.internal.util.ActionSubscriber.onError(ActionSubscriber.java:44)
     at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:157)
     at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120)
     at rx.exceptions.Exceptions.throwOrReport(Exceptions.java:204)
     at rx.observers.SafeSubscriber.onNext(SafeSubscriber.java:144)
     at rx.internal.operators.OperatorOnErrorResumeNextViaFunction$4.onNext(OperatorOnErrorResumeNextViaFunction.java:153)
     at rx.internal.operators.OperatorMap$MapSubscriber.onNext(OperatorMap.java:74)
     at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:391)
     at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:353)
     at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:838)
     at rx.observers.Subscribers$5.onNext(Subscribers.java:229)
     at rx.internal.operators.OperatorZip$Zip.tick(OperatorZip.java:264)
     at rx.internal.operators.OperatorZip$Zip$InnerSubscriber.onNext(OperatorZip.java:335)
     at rx.internal.operators.OperatorMap$MapSubscriber.onNext(OperatorMap.java:74)
     at rx.internal.util.ScalarSynchronousObservable$WeakSingleProducer.request(ScalarSynchronousObservable.java:268)
     at rx.Subscriber.setProducer(Subscriber.java:211)
     at rx.internal.operators.OperatorMap$MapSubscriber.setProducer(OperatorMap.java:99)
     at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:79)
     at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:75)
     at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:50)
     at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)
     at rx.Observable.unsafeSubscribe(Observable.java:8452)
     at rx.internal.operators.OperatorZip$Zip.start(OperatorZip.java:214)
     at rx.internal.operators.OperatorZip$ZipSubscriber.onNext(OperatorZip.java:156)
     at rx.internal.operators.OperatorZip$ZipSubscriber.onNext(OperatorZip.java:122)
     at rx.internal.util.ScalarSynchronousObservable$WeakSingleProducer.request(ScalarSynchronousObservable.java:268)
     at rx.Subscriber.setProducer(Subscriber.java:209)
     at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:79)
     at rx.internal.util.ScalarSynchronousObservable$1.call(ScalarSynchronousObservable.java:75)
     at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:50)
     at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)
     at rx.Observable.unsafeSubscribe(Observable.java:8452)
     at rx.internal.util.ScalarSynchronousObservable$4.call(ScalarSynchronousObservable.java:227)
     at rx.internal.util.ScalarSynchronousObservable$4.call(ScalarSynchronousObservable.java:220)
     at rx.Observable.unsafeSubscribe(Observable.java:8452)
     at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:248)

Answer 1

您可以使用onErrorResumeNext 运算符。示例用法：

fun BaseFragment.callGallery(view: View){
    view
      .clicks()
      .bindToLifecycle(this)
      .onErrorResumeNext { err -> log(err); Observable.empty() }
      .subscribe {
         // do stuff
      }
}

Answer 2

RxJava 有Plugins 可以很好地支持横切关注点。特别是 RxJavaErrorHandler 被设计为附加全局错误处理行为，如下所示：

RxJavaPlugins.getInstance().registerErrorHandler(object : RxJavaErrorHandler() {
    override fun handleError(e: Throwable?) {
        println("Global error handler: $e")
    }
})

Observable.just(1).concatMap({ Observable.error<Int>(Exception("Just throwing $it")) })
        .subscribe({
            println("I'll not be called")
        }, {
            println("Specific error handler: $it")
        })

上面的代码会打印：

Error occurred java.lang.Exception: Just throwing 1
A default error handler: java.lang.Exception: Just throwing 1

我怀疑当您确实遇到未处理的错误时，您也会对改进堆栈跟踪中可用的诊断信息感兴趣。为此，有一个RxJavaStackTracer，当使用RxJavaPlugins.getInstance().registerObservableExecutionHook(RxJavaStackTracer()) 时会增强堆栈跟踪信息。 github 上的Stacktraces and subscribeOn/observeOn issue 是该主题的好读物。