Rxjava interval 与flatmap结合异常处理

通常在Rxjava中我会用interval当定时器与flatmap结合发起网络请求.在项目中的实际需求是每两个小时请求一次,这条异常就跳过,继续请求下一条,但是发现一个问题就是请求可能会超时异常.为了举例方便,现在我将代码改为类似的可能返回异常的简单替代

 public static void main(String[] args) {
        Observable.interval(0, 1, TimeUnit.SECONDS, Schedulers.trampoline())
                .flatMap(new Function<Long, ObservableSource<Long>>() {
                    @Override
                    public ObservableSource<Long> apply(Long aLong) throws Exception {
                        return Observable.just(aLong == 3 ? aLong / 0 : aLong * 2);
                    }
                })
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long integer) throws Exception {
                        System.out.println(integer + "");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        System.out.println(throwable.toString());
                    }
                });

    }

我们的期望是出错调用错误处理后interval继续发射数据,继续请求下一个,我们看一打印结果:

0
2
4
java.lang.ArithmeticException: / by zero

结果是显而易见的,在异常出现且被处理后,interval停止了数据的发射,究其原因是在整个链路中flatMap返回了一个Observable<Throwable>导致链路中断,让interval继续轮询请求的关键便是flatMap返回异常时链路不能被中断,但是实在找不到合适的操作符,于是干脆把flatMap的操作放在了订阅者中,这样interval就不会被flatMap返回的错误结果打断了!先贴代码:

     Observable.interval(0, 1, TimeUnit.SECONDS, Schedulers.trampoline())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Observable.just(aLong).
                                flatMap(new Function<Long, ObservableSource<Long>>() {
                                    @Override
                                    public ObservableSource<Long> apply(Long aLong) throws Exception {
                                        return Observable.just(aLong == 3 ? aLong / 0 : aLong * 2);
                                    }
                                })
                                .subscribe(new Consumer<Long>() {
                                    @Override
                                    public void accept(Long integer) throws Exception {
                                        System.out.println(integer + "");
                                    }
                                }, new Consumer<Throwable>() {
                                    @Override
                                    public void accept(Throwable throwable) throws Exception {
                                        System.out.println(throwable.toString());
                                    }
                                });
                    }
                });

打印结果:可以继续下去了

0
2
4
java.lang.ArithmeticException: / by zero
8
10
12

但是这样真的好不优雅,求大神指教有什么优雅的方法吗?