使用RxJava for Android重试observable，直到满足某些条件

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        // 发射需要进行重试的事件流
        emitter.onNext(1);
        emitter.onNext(2);
        emitter.onNext(3);
        emitter.onError(new Exception("Error"));
    }
})
.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
    @Override
    public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
        // 定义重试的条件
        return throwableObservable
                .zipWith(Observable.range(1, 3), new BiFunction<Throwable, Integer, Integer>() {
                    @Override
                    public Integer apply(Throwable throwable, Integer retryCount) throws Exception {
                        // 控制重试的次数
                        if (retryCount <= 3) {
                            return retryCount;
                        } else {
                            throw Exceptions.propagate(throwable);
                        }
                    }
                })
                .flatMap(new Function<Integer, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Integer retryCount) throws Exception {
                        // 控制重试的时间间隔
                        return Observable.timer(retryCount, TimeUnit.SECONDS);
                    }
                });
    }
})
.subscribe(new Observer<Integer>() {
    @Override
    public void onSubscribe(Disposable d) {
        // 订阅时的操作
    }

    @Override
    public void onNext(Integer integer) {
        // 处理每个事件的操作
    }

    @Override
    public void onError(Throwable e) {
        // 处理错误的操作
    }

    @Override
    public void onComplete() {
        // 完成时的操作
    }
});