rxjava实用示例

每隔两秒执行一次

Observable.interval(2, 2, TimeUnit.SECONDS).subscribe(new Action1<Long>() {
        @Override
        public void call(Long aLong) {
            //TODO WHAT YOU WANT
        }
    });

在两秒后去执行一些操作（比如启动页跳转到主页面）

Observable.timer(2, TimeUnit.SECONDS).subscribe(new Action1<Long>() {
        @Override
        public void call(Long aLong) {
            //TODO WHAT YOU WANT
        }
    });

模拟发送验证码,count秒钟后可再次点击

Observable.interval(0, 1, TimeUnit.SECONDS)
            .take(count + 1)
            .map(new Function<Long, Long>() {
                @Override
                public Long apply(@NonNull Long aLong) throws Exception {
                    return count - aLong;
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
             //订阅时执行
            .doOnSubscribe(new Consumer<Disposable>() {
                @Override
                public void accept(@NonNull Disposable disposable) throws Exception {
                    button.setEnabled(false);
                }
            })
            .subscribe(new Observer<Long>() {
                @Override
                public void onSubscribe(Disposable d) {}
                @Override
                public void onNext(Long aLong) {
                    button.setText("剩余" + aLong + "秒");
                }
                @Override
                public void onError(Throwable e) {}
                @Override
                public void onComplete() {
                    //恢复状态
                    button.setEnabled(true);
                    button.setText("发送验证码");
                }
            });

登录框按钮状态判断

    Observable<CharSequence> ObservableName = RxTextView.textChanges(mEtPhone);
    Observable<CharSequence> ObservablePassword = RxTextView.textChanges(mEtPassword);

    Observable.combineLatest(ObservableName, ObservablePassword, new Func2<CharSequence, CharSequence, Boolean>() {
        @Override
        public Boolean call(CharSequence phone, CharSequence password) {
            return isPhoneValid(phone.toString()) && isPasswordValid(password.toString());
        }
    }).subscribe(new Action1<Boolean>() {
        @Override
        public void call(Boolean aBoolean) {
            RxView.enabled(mBtLogin).call(aBoolean);
        }
    });

缓存策略

    //内存，磁盘，网络有一个有值即返回
    Observable source = Observable
    .concat(memory, disk, network)
    .first(new Func1() {
      @Override public Boolean call(Data data) {
      return data.isUpToDate();
    }
});

合并两个数据源场景

    //合并本地及网络数据，如离线加入购物车等 
    Observable.merge(getLocal(), getNet())  
       .observeOn(AndroidSchedulers.mainThread())  
          .subscribe(new Subscriber<String>() {  
              @Override  
              public void onCompleted() {  
                 Log.d(TAG, "onCompleted: onCompleted");
             }  

             @Override  
              public void onError(Throwable e) {  
                Log.d(TAG, "onError: onError");
              }  

              @Override  
              public void onNext(String data) {  
                   Log.d(TAG, "onNext: only one ! ");
         });

把多个网络结果合并成一个

Observable<UserAndEvents> combined = Observable.zip(userObservable, eventsObservable, new Func2<JsonObject, JsonArray, UserAndEvents>() {
  @Override
  public UserAndEvents call(JsonObject jsonObject, JsonArray jsonElements) {
    return new UserAndEvents(jsonObject, jsonElements);
  }
});

快速切换线程

Observable.Transformer schedulersTransformer() {
        return new Observable.Transformer() {
            @Override
            public Object call(Object observable) {
                return ((Observable)  observable).subscribeOn(Schedulers.io())
                        .unsubscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }
observable.compose (schedulersTransformer()).subscribe(subscriber)
//上面的等价于下面的
observable.subscribeOn(Schedulers.io())
.unsubscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(subscriber);

使用schedulePeriodically做轮询请求

    Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(final Subscriber<? super String> observer) {

            Schedulers.newThread().createWorker()
                  .schedulePeriodically(new Action0() {
                      @Override
                      public void call() {
                          observer.onNext(doNetworkCallAndGetStringResult());
                      }
                  }, INITIAL_DELAY, POLLING_INTERVAL, TimeUnit.MILLISECONDS);
        }
    }).subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
            log.d("polling….”));
        }
    })

RxBinding

防止一个按钮重复点击

RxView.clicks(button).debounce(300, TimeUnit.MILLISECONDS).subscribe(new Action1<Void>() {
        @Override
        public void call(Void aVoid) {
            Log.i("test", "clicked");
        }
    });
    //throttleFirst与throttleLast/sample不同，在每个采样周期内，它总是发射原始Observable的第一项数据，而不是最近的一项。
    RxView.clicks(button).throttleFirst(300, TimeUnit.SECONDS)
            .subscribe(new Observer<Object>() {
                @Override
                public void onSubscribe(Disposable d) {}
                @Override
                public void onNext(Object o) {
                    System.out.println("俊俊俊点击了按钮");
                }
                @Override
                public void onError(Throwable e) {}
                @Override
                public void onComplete() {}
            });

优化网络搜索

RxTextView.textChanges(edittext)
            //当你敲完字之后停下来的半秒就会执行下面语句
            .debounce(500, TimeUnit.MILLISECONDS)
            //下面这两个都是数据转换
            //flatMap：当同时多个网络请求访问的时候，前面的网络数据会覆盖后面的网络数据
            //switchMap：当同时多个网络请求访问的时候，会以最后一个发送请求为准，前面网路数据会被最后一个覆盖
            .switchMap(new Function<CharSequence, ObservableSource<List<String>>>() {
                @Override
                public ObservableSource<List<String>> apply(@NonNull CharSequence charSequence) throws Exception {
                    //网络操作，获取我们需要的数据
                    List<String> list = new ArrayList<String>();
                    list.add("2017年款最新帅哥俊俊俊");
                    list.add("找不到2017年比俊俊俊更帅的人");
                    return Observable.just(list);
                }
            })
            //网络请求是在子线程的
            .subscribeOn(Schedulers.io())
            //界面更新在主线程
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<List<String>>() {
                @Override
                public void accept(@NonNull List<String> strings) throws Exception {
                    //界面更新，这里用打印替代
                    System.out.println(strings.toString());
                }
            });

RxPermissions

请求相机权限

RxPermissions.getInstance(this)
.request(Manifest.permission.CAMERA)
.subscribe(granted -> {
    if (granted) { // 用户同意了（在6.0之前的手机始终都为true）
      //可以拍照了
    } else {
       //可以在这里提示用户，或者再次请求
    }
});