RxJava学习(十二)

操作符

结合操作

zip和zipWith

zip

Observable<String> observable = Observable.just("10");
        Observable<String> observable2 = Observable.just("20");
        Observable.zip(observable, observable2, new Func2<String, String, Integer>() {
            @Override
            public Integer call(String s, String s2) {
                return Integer.parseInt(s)+Integer.parseInt(s2);
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {

            }
            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer s) {
                Log.d(TAG, "onNext: "+s);
            }
        });

zipWith:

Observable<String> observable = Observable.just("20 ");
        Observable.just("10").zipWith(observable, new Func2<String, String, Integer>() {
            @Override
            public Integer call(String s, String s2) {
                return null;
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onNext(Integer s) {
            }
        });

区别：
1.zip是Observable的静态方法，而zipWith是对象的方法
2.可以实现同样的效果，
3.zip可以接收多个Observable，zipWith只能连接两个Observable。

range
连续发射一个范围内的数据，下面的代码输出0，1，2

Observable.range(0,3).subscribe(new Action1<Integer>() {
           @Override
           public void call(Integer integer) {
               Log.d(TAG, "call: "+integer);
           }
       });

第一个参数为开始位置，第二个参数为数量。

异常重试操作

retry

getMoiveInfo.retry(new Func2<Integer, Throwable, Boolean>() {
         @Override
         public Boolean call(Integer integer, Throwable throwable) {
             return throwable instanceof RuntimeException && integer < 3;
         }
     }).

retry可以接收一个判定条件，当满足这个条件的情况下，会重新执行前面的网络请求，直到不满足条件，才执行onError处理异常，

retryWhen

retryWhen和retry类似，区别是，retryWhen将onError中的Throwable传递给一个函数，这个函数产生另一个Observable，retryWhen观察它的结果再决定是不是要重新订阅原始的Observable。如果这个Observable发射了一项数据，它就重新订阅，如果这个Observable发射的是onError通知，它就将这个通知传递给观察者然后终止。

下面引用一个例子
获取用户信息

userApi.getUserInfoNoToken()
                        //总共重试3次，重试间隔3000毫秒
                        .retryWhen(new RetryWithDelay(3, 3000))
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribeOn(Schedulers.io())
                        .subscribe(new Action1<Response>() {
                            @Override
                            public void call(Response response) {
                                String content = new String(((TypedByteArray) response.getBody()).getBytes());
                                printLog(tvLogs, "", content);
                            }
                        }, new Action1<Throwable>() {
                            @Override
                            public void call(Throwable throwable) {
                                throwable.printStackTrace();
                            }
                        });

public class RetryWithDelay implements
            Func1<Observable<? extends Throwable>, Observable<?>> {

        private final int maxRetries;
        private final int retryDelayMillis;
        private int retryCount;

        public RetryWithDelay(int maxRetries, int retryDelayMillis) {
            this.maxRetries = maxRetries;
            this.retryDelayMillis = retryDelayMillis;
        }

        @Override
        public Observable<?> call(Observable<? extends Throwable> attempts) {
            return attempts
                    .flatMap(new Func1<Throwable, Observable<?>>() {
                        @Override
                        public Observable<?> call(Throwable throwable) {
                            if (++retryCount <= maxRetries) {
                                // When this Observable calls onNext, the original Observable will be retried (i.e. re-subscribed).
                                printLog(tvLogs, "", "get error, it will try after " + retryDelayMillis
                                        + " millisecond, retry count " + retryCount);
                                return Observable.timer(retryDelayMillis,
                                        TimeUnit.MILLISECONDS);
                            }
                            // Max retries hit. Just pass the error along.
                            return Observable.error(throwable);
                        }
                    });
        }
    }

定时操作
timer和delay

timer:Observable的静态方法，如果不需要根据结果做操作，单纯实现定时任务，采用这个

Observable.timer(4,TimeUnit.SECONDS).subscribe(new Action1<Long>() {
            @Override
            public void call(Long aLong) {
                //4秒后启动新的页面
                startActivity(new Intent(MainActivity.this,SecondActivity.class));
            }
        });

delay:对象的方法

Observable.just("10").delay(4, TimeUnit.SECONDS).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                startActivity(new Intent(MainActivity.this,SecondActivity.class));
            }
        });

Schedulers.newThread().createWorker().schedule(new Action0() {
            @Override
            public void call() {
                startActivity(new Intent(MainActivity.this,SecondActivity.class));
            }
        },4, TimeUnit.SECONDS);

上面三种方式都实现了4秒后执行操作，具体采用哪个看具体需求。

周期性操作

Schedulers.newThread().createWorker().schedulePeriodically(new Action0() {
          @Override
          public void call() {
              Log.d(TAG, "call: 执行了");
          }
      },0,4,TimeUnit.SECONDS);

上面实现了每4秒执行一次操作。