java - onErrorResumeNext 后间隔继续工作

Question

我正在使用 Interval 运算符，即使我的管道发生异常，我也想继续发射项目。

所以我尝试使用 onErrorResumeNext 在出现异常时发出一个项目。但是我看到在发出这个项目之后，间隔停止发出更多的项目。

这是我的单元测试。

@Test
public void testIntervalObservableWithError() {
    Subscription subscription = Observable.interval(50, TimeUnit.MILLISECONDS)
            .map(time -> "item\n")
            .map(item -> item = null)
            .map(String::toString)
            .onErrorResumeNext(t-> Observable.just("item with error emitted"))
            .subscribe(System.out::print, t->{
                        System.out.println(t);
                    }
                   );
    TestSubscriber testSubscriber = new TestSubscriber((Observer) subscription);
    testSubscriber.awaitTerminalEvent(20000, TimeUnit.MILLISECONDS);
}

我对这种行为感到困惑，为什么 observable 在从 onErrorResumeNext

接收到一个项目时取消订阅

解决方案:

经过一些解释，我意识到当错误发生时，可观察到的 t's 完成了。所以我最终将可能有异常的可观察对象包装到另一个可观察对象中，并且我使用的是 flatMap。因此，主 Observable 会继续发射项目。

@Test
public void testIntervalObservableWithError() {
    Observable.interval(100, TimeUnit.MILLISECONDS)
            .map(time -> "item\n")
            .flatMap(item -> Observable.just(item)
                    .map(String::toString))
            .subscribe(System.out::print); 
    TestSubscriber testSubscriber = new TestSubscriber();
    testSubscriber.awaitTerminalEvent(5000, TimeUnit.MILLISECONDS);
}

如果有任何运算符可以完成所有这些魔术，我想知道。

注册

Answer 1

您的订阅中断是因为当 onErrorResumeNext 被触发时，您的上游已经完成并出现错误。你只需发出一个项目而不是让异常进入下游。为了保持上游 Activity ，您必须防止在其上抛出异常。

对于您的特定示例解决方案可以是这样的:

...
    .map(time -> "item\n")
    .map(item -> item = null)
    .map(item -> {
        try {
            return item.toString();
        } catch (NullPointerException e) {
            return "item with error emitted";
    })
    //no onErrorResumeNext()
    .subscribe ...

onErrorResumeNext 只是用一个项目替换错误并调用 onComplete。