java - Observable的doOnError正确位置

Question

我对Observers有点陌生，我仍在努力弄清楚它们。我有以下代码:

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的位置重要吗？另外，在这段代码中添加多个调用会产生什么效果？我意识到我可以做到这一点，并且所有这些都被调用，但我不确定它的目的是什么。

Answer 1

是的，确实如此。 doOnError 当错误在该特定点通过流时起作用，因此如果 doOnError throw(s) 之前的运算符，将调用您的操作。但是，如果将 doOnError 进一步放置，它可能会也可能不会被调用，具体取决于链中的下游运算符。

给定

Observer<Object> ignore = new Observer<Object>() {
    @Override public void onCompleted() {
    }
    @Override public void onError(Throwable e) {
    }
    @Override public void onNext(Object t) {
    }
};

例如，以下代码将始终调用 doOnError:

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);

大多数运算符(operator)都会反弹源自下游的异常。

如果通过 onErrorResumeNext 或 onExceptionResumeNext 转换异常，则添加多个 doOnError 是可行的:

Observable.<Object>error(new RuntimeException())
.doOnError(e -> log(e))
.onErrorResumeNext(Observable.<Object>error(new IllegalStateException()))
.doOnError(e -> log(e)).subscribe(ignore);

否则，您将在链的多个位置记录相同的异常。