我对观察者来说是个新手,我仍在努力解决这些问题.我有以下代码：
observableKafka.getRealTimeEvents()
.filter(this::isTrackedAccount)
.filter(e -> LedgerMapper.isDepositOrClosedTrade((Transaction) e.getPayload()))
.map(ledgerMapper::mapLedgerTransaction)
.map(offerCache::addTransaction)
.filter(offer -> offer != null) // Offer may have been removed from cache since last check
.filter(Offer::isReady)
.doOnError(throwable -> {
LOG.info("Exception thrown on realtime events");
})
.forEach(awardChecker::awardFailOrIgnore);
getRealTimeEvents()返回一个Observable< Event>.
.doOnError的位置是否重要？另外,在这段代码中添加多个调用会产生什么影响？我已经意识到我可以做到并且所有这些都被调用,但我不确定它的目的是什么.
解决方法
是的,它确实.当错误在特定点传递流时,doOnError会起作用,因此如果doOnError之前的操作符抛出,则会调用您的操作.但是,如果进一步放置doOnError,可能会也可能不会调用它,具体取决于链中的下游运算符.
特定
Observer<Object> ignore = new Observer<Object>() {
@Override public void onCompleted() {
}
@Override public void onError(Throwable e) {
}
@Override public void onNext(Object t) {
}
};
Observable.<Object>error(new Exception()).doOnError(e -> log(e)).subscribe(ignore);
但是,此代码不会：
Observable.just(1).doOnError(e -> log(e)) .flatMap(v -> Observable.<Integer>error(new Exception())).subscribe(ignore);
大多数操作符都会反弹源自下游的异常.
如果通过onErrorResumeNext或onExceptionResumeNext转换异常,则添加multipe doOnError是可行的：
Observable.<Object>error(new RuntimeException()) .doOnError(e -> log(e)) .onErrorResumeNext(Observable.<Object>error(new IllegalStateException())) .doOnError(e -> log(e)).subscribe(ignore);
否则,您将在链的多个位置记录相同的异常.
