RxJava 日常使用

本文转载自：

http://blog.csdn.net/theone10211024/article/details/50435325

 

1、Scheduler线程切换

这种场景常常会在“后台线程取数据，主线程展现”的模式中看见

 

Observable.just(1, 2, 3, 4)
            .subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
            .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
            .subscribe(new Action1<Integer>() {
        @Override
        public void call(Integer number) {
            Log.d(tag, "number:" + number);
        }
    });

 

 

 

2、使用debounce作textSearch

用简单的话讲就是当N个结点发生的时间太靠近（即发生的时间差小于设定的值T），debounce就会自动过滤掉前N-1个结点。

好比在作百度地址联想的时候，可使用debounce减小频繁的网络请求。避免每输入（删除）一个字就作一次联想

 

RxTextView.textChangeEvents(inputEditText)
      .debounce(400, TimeUnit.MILLISECONDS)
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new Observer<TextViewTextChangeEvent>() {
    @Override
    public void onCompleted() {
        log.d("onComplete");
    }

    @Override
    public void onError(Throwable e) {
        log.d("Error");
    }

    @Override
    public void onNext(TextViewTextChangeEvent onTextChangeEvent) {
        log.d(format("Searching for %s", onTextChangeEvent.text().toString()));
    }
});

 

 

3、Retrofit结合RxJava作网络请求框架

这里不做详解，具体的介绍能够看扔物线的这篇文章，对RxJava的入门者有很大的启发。其中也讲到了RxJava和Retrofit如何结合来实现更简洁的代码

 

 

4、RxJava代替EventBus进行数据传递：RxBus

注意：RxBus并非一个库，而是一种模式，是使用了RxJava的思想来达到EventBus的数据传递效果。这篇文章把RxBus讲的比较详细。

5、使用combineLatest合并最近N个结点

例如：注册的时候全部输入信息（邮箱、密码、电话号码等）合法才点亮注册按钮。

Observable<CharSequence> _emailChangeObservable = RxTextView.textChanges(_email).skip(1);
Observable<CharSequence> _passwordChangeObservable = RxTextView.textChanges(_password).skip(1);
Observable<CharSequence>   _numberChangeObservable = RxTextView.textChanges(_number).skip(1);

Observable.combineLatest(_emailChangeObservable,
              _passwordChangeObservable,
              _numberChangeObservable,
              new Func3<CharSequence, CharSequence, CharSequence, Boolean>() {
                  @Override
                  public Boolean call(CharSequence newEmail,
                                      CharSequence newPassword,
                                      CharSequence newNumber) {

                      Log.d("xiayong",newEmail+" "+newPassword+" "+newNumber);
                      boolean emailValid = !isEmpty(newEmail) &&
                                           EMAIL_ADDRESS.matcher(newEmail).matches();
                      if (!emailValid) {
                          _email.setError("Invalid Email!");
                      }

                      boolean passValid = !isEmpty(newPassword) && newPassword.length() > 8;
                      if (!passValid) {
                          _password.setError("Invalid Password!");
                      }

                      boolean numValid = !isEmpty(newNumber);
                      if (numValid) {
                          int num = Integer.parseInt(newNumber.toString());
                          numValid = num > 0 && num <= 100;
                      }
                      if (!numValid) {
                          _number.setError("Invalid Number!");
                      }

                      return emailValid && passValid && numValid;

                  }
              })//
              .subscribe(new Observer<Boolean>() {
                  @Override
                  public void onCompleted() {
                      log.d("completed");
                  }

                  @Override
                  public void onError(Throwable e) {
                     log.d("Error");
                  }

                  @Override
                  public void onNext(Boolean formValid) {
                     _btnValidIndicator.setEnabled(formValid);
                  }
              });

6、使用merge合并两个数据源。

例如一组数据来自网络，一组数据来自文件，须要合并两组数据一块儿展现。

 

Observable.merge(getDataFromFile(), getDataFromNet())
              .observeOn(AndroidSchedulers.mainThread())
              .subscribe(new Subscriber<String>() {
                  @Override
                  public void onCompleted() {
                      log.d("done loading all data");
                  }

                  @Override
                  public void onError(Throwable e) {
                      log.d("error");
                  }

                  @Override
                  public void onNext(String data) {
                      log.d("all merged data will pass here one by one!")
              });

 

7、使用concat和first作缓存

  依次检查memory、disk和network中是否存在数据，任何一步一旦发现数据后面的操做都不执行。

 

Observable<String> memory = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        if (memoryCache != null) {
            subscriber.onNext(memoryCache);
        } else {
            subscriber.onCompleted();
        }
    }
});
Observable<String> disk = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        String cachePref = rxPreferences.getString("cache").get();
        if (!TextUtils.isEmpty(cachePref)) {
            subscriber.onNext(cachePref);
        } else {
            subscriber.onCompleted();
        }
    }
});

Observable<String> network = Observable.just("network");

//依次检查memory、disk、network
Observable.concat(memory, disk, network)
.first()
.subscribeOn(Schedulers.newThread())
.subscribe(s -> {
    memoryCache = "memory";
    System.out.println("--------------subscribe: " + s);
});

 

8、使用timer作定时操做。当有“x秒后执行y操做”相似的需求的时候，想到使用timer

例如：2秒后输出日志“hello world”，而后结束。

 

Observable.timer(2, TimeUnit.SECONDS)
              .subscribe(new Observer<Long>() {
                  @Override
                  public void onCompleted() {
                      log.d ("completed");
                  }

                  @Override
                  public void onError(Throwable e) {
                      log.e("error");
                  }

                  @Override
                  public void onNext(Long number) {
                      log.d ("hello world");
                  }
              });

 

 

9、使用interval作周期性操做。当有“每隔xx秒后执行yy操做”相似的需求的时候，想到使用interval

例如：每隔2秒输出日志“helloworld”。

 

Observable.interval(2, TimeUnit.SECONDS)
         .subscribe(new Observer<Long>() {
             @Override
             public void onCompleted() {
                log.d ("completed");
             }

             @Override
             public void onError(Throwable e) {
                log.e("error");
             }

             @Override
             public void onNext(Long number) {
                log.d ("hello world");
             }
         });

 

 

10、使用throttleFirst防止按钮重复点击

ps：debounce也能达到一样的效果

 

RxView.clicks(button)
              .throttleFirst(1, TimeUnit.SECONDS)
              .subscribe(new Observer<Object>() {
                  @Override
                  public void onCompleted() {
                        log.d ("completed");
                  }

                  @Override
                  public void onError(Throwable e) {
                        log.e("error");
                  }

                  @Override
                  public void onNext(Object o) {
                       log.d("button clicked");
                  }
              });

 

11、使用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….”));
            }
        })

12、RxJava进行数组、list的遍历

 

String[] names = {"Tom", "Lily", "Alisa", "Sheldon", "Bill"};
Observable
        .from(names)
        .subscribe(new Action1<String>() {
            @Override
            public void call(String name) {
                log.d(name);
            }
        });

 

十3、解决嵌套回调（callback hell）问题

 

NetworkService.getToken("username", "password")
    .flatMap(s -> NetworkService.getMessage(s))
    .subscribe(s -> {
        System.out.println("message: " + s);
})

 

 

十4、响应式的界面

好比勾选了某个checkbox，自动更新对应的preference

 

SharedPreferences preferences = PreferenceManager.getDefaultSharedPreferences(this);
RxSharedPreferences rxPreferences = RxSharedPreferences.create(preferences);

Preference<Boolean> checked = rxPreferences.getBoolean("checked", true);

CheckBox checkBox = (CheckBox) findViewById(R.id.cb_test);
RxCompoundButton.checkedChanges(checkBox)
        .subscribe(checked.asAction());