android - RxJava RetryWhen，这个处理器只允许一个订阅者

Question

我正在学习如何在 RxJava2 中进行数据轮询

到目前为止，这是我的代码。

 private io.reactivex.Single<String> getMyTask() {
        return io.reactivex.Single.fromCallable(new Callable<String>() {
            @Override
            public String call() throws Exception {
                Log.d("ERSEN","Task Started!");
                Random random = new Random(System.currentTimeMillis());

                if(random.nextBoolean()){
                    return "WORK COMPLETED";
                }
                Log.d("ERSEN","Task Had An Error!");
                throw new IllegalArgumentException();
            }
        });
    }

上面是我的Single，它发出一个String，基本上是在模拟一些工作。我还让任务随机成功，并在轮询事件无法检查重新订阅是否正确发生时测试失败

我的问题

 compositeDisposable.add(getMyTask()
                .repeatWhen(new Function<Flowable<Object>, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(final Flowable<Object> objectFlowable) throws Exception {
                        return objectFlowable.delay(INTERVAL, TimeUnit.SECONDS);
                    }
                })
                .retryWhen(throwableFlowable -> throwableFlowable.flatMap(new Function<Throwable, Publisher<?>>() {
                    @Override
                    public Publisher<?> apply(Throwable throwable) throws Exception {
                        if (throwable instanceof ClassCastException) {
                            return Flowable.error(throwable);
                        }
                        return throwableFlowable.delay(INTERVAL, TimeUnit.SECONDS);
                    }
                }))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(this::onSuccess, this::onError));

在上面，当 Observable 成功发出一些数据时，我重新订阅了它。我在使用 retryWhen 时遇到问题。

对于这个例子，我不希望在发生 ClassCastException 时重试。在我的 Observable 中，这没有产生，这是有原因的，因为我正在测试仅在某些错误上重试的逻辑

但是，当 Observable 产生错误时，我用上面的代码恢复了这个错误

This processor allows only a single Subscriber

我不确定哪里出了问题，我一直在关注这篇博文

http://blog.danlew.net/2016/01/25/rxjavas-repeatwhen-and-retrywhen-explained/

感谢阅读让我知道您是否希望我发布更多详细信息

Answer 1

您正在重新订阅 retryWhen 中的错误流，这是不允许的，在您的情况下没有意义。您应该在 flatMap 中延迟一个值:

    .retryWhen(throwableFlowable -> throwableFlowable.flatMap(
        new Function<Throwable, Publisher<?>>() {
            @Override
            public Publisher<?> apply(Throwable throwable) throws Exception {
                if (throwable instanceof ClassCastException) {
                    return Flowable.error(throwable);
                }
                return Flowable.just("ignored").delay(INTERVAL, TimeUnit.SECONDS);
            }
        }
    ))