在RetryWhen()上输入错误并带有延迟

Question

我正在学习本教程,然后添加以下行:

.retryWhen(errors -> errors.flatMap(error -> Observable.timer(30, TimeUnit.SECONDS)))

对我来说Transformer,但它会触发编译错误:

错误:不兼容的类型:无法推断类型变量R(参数不匹配; lambda表达式中的错误返回类型Observable无法转换为Publisher)其中R,T是类型变量:R extends方法flatMap中声明的Object(函数> )T扩展在类Flowable中声明的Object

而下面的波浪线error -> Observable.timer(30, TimeUnit.SECONDS)说:

没有类型变量R的实例存在,以便Observable符合Publisher

我究竟做错了什么？这个例子中缺少什么？

Answer 1

我究竟做错了什么？

请检查运营商的签名,以便使用正确的类型:https://github.com/ReactiveX/RxJava#base-class-vs-base-type

JavaDoc:

public final Single<T> retryWhen(
    Function<? super Flowable<Throwable>,? extends Publisher<?>> handler)

因为本教程使用Observable.timer而没有任何问题.

该教程早于RxJava 2.事实上,上面链接的Javadoc包含一个示例Flowable.timer():

Single.timer(1, TimeUnit.SECONDS)
.doOnSubscribe(s -> System.out.println("subscribing"))
.map(v -> { throw new RuntimeException(); })
.retryWhen(errors -> {
    AtomicInteger counter = new AtomicInteger();
    return errors
              .takeWhile(e -> counter.getAndIncrement() != 3)
              .flatMap(e -> {
                  System.out.println("delay retry by " + counter.get() + " second(s)");
//                vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv
                  return Flowable.timer(counter.get(), TimeUnit.SECONDS);
//                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
              });
})
.blockingGet();

这个东西只与Single有关

retryWhen和repeatWhen使用Publisher作为重做信号设计,使我们可以利用背压一个这样重做信号在同一时间只问.有了Observable,处理程序有可能只是立即转储大量信号,操作员可能会出现意外行为.