java – Observable的doOnError正确位置

我对观察者来说是个新手,我仍在努力解决这些问题.我有以下代码：

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

例如,以下代码将始终调用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);

大多数操作符都会反弹源自下游的异常.

如果通过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);

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

原文地址：https://www.jb51.cc/java/126788.html