rxjava

Posted 星耀长河

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了rxjava相关的知识，希望对你有一定的参考价值。

前言：阳光总在风雨后，人生没有白吃的苦，也没有白走的路，要知道，你现在吃的苦，现在受的伤，会成为日后的徽章。

一、概述

提高开发效果，降低维护成本一直是团队追求的宗旨。现在安卓里面RxJava2+Retrofit+OKHttp的网络请求框架是最流行的，几乎一统江湖，配合响应式式编程的RxJava使用也越来越多。如果英文比较好而且有兴趣的可以到官网学习：ReactiveX的官方网站。（源码和其他相关链接在文章最后给出）

RxJava到底是什么？

      RxJava在GitHub上《RxJava》的自我介绍是：a library for composing asynchronous and event-based programs using observable sequences for the Java VM.(一个在Java VM 上使用可观测的序列来组成异步的，基于事件的程序的库)，有些人可能感到疑惑，其实本质上可以用一词来概括——“异步”，它就是一个异步的操作库，而别的定语都基于这之上。

RxJava的好处是啥？

同样是异步操作，为什么不用Handler或者AsyncTask等异步机制？

因为：简洁

异步操作很关键的一点是简洁性，因为在调度过程比较复杂的情况下，异步代码既难写也难读懂。android创造的Handler和AsyncTask其实都是为了让异步代码更加简洁，RxJava的优势也是简洁，它的特别之处是：随着程序代码逻辑的变得越来越复杂，它依然保持简洁。

二、使用详解

1.基本实现

RxJava最核心的两个东西Observable（被观察者、事件源）和Observer（观察者），Observable发出一系列的事件，Observer处理这些事件。在Observer接收到事件处理之前我们很方便地对结果做出各种拦截处理等。

RxJava的流程图大致如下：

1).添加依赖库

在model的build.gradle文件内添加RxJava2依赖库，注意RxJava1与RxJava2依赖不能共存。

  1.  implementation \'io.reactivex.rxjava2:rxjava:2.0.4\'
  2.  implementation \'io.reactivex.rxjava2:rxandroid:2.0.1\'

2).三步创建实例

  1.  //1、创建被观察者Observable
  2.  Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
  3.  @Override
  4.  public void subscribe(ObservableEmitter<String> e) throws Exception {
  5.  e.onNext("RxJava：e.onNext== 第一次");
  6.  e.onNext("RxJava：e.onNext== 第二次");
  7.  e.onNext("RxJava：e.onNext== 第三次");
  8.  e.onComplete();
  9.  }
  10.  });
  11.  
  12.  //2、创建观察者Observer
  13.  Observer<String> observer = new Observer<String>() {
  14.  @Override
  15.  public void onSubscribe(Disposable d) {
  16.  Log.e(TAG, "onSubscribe == 订阅");
  17.  }
  18.  
  19.  @Override
  20.  public void onNext(String s) {
  21.  Log.e(TAG, "onNext == " + s);
  22.  }
  23.  
  24.  @Override
  25.  public void onError(Throwable e) {
  26.  Log.e(TAG, "onError == " + e.getMessage());
  27.  }
  28.  
  29.  @Override
  30.  public void onComplete() {
  31.  Log.e(TAG, "onComplete == ");
  32.  }
  33.  };
  34.  
  35.  //3、订阅(观察者观察被观察者)
  36.  observable.subscribe(observer);

直接运行上面的例子，打印结果如下：

上述例子中，通过new Observer<T>()创建观察者，实现相应的方法，Observable.create(ObservableOnSubscribe)创建被观察者，subscribe()将两者关联起来。

3).观察者模式

RxJava的异步实现，是通过一种拓展的观察者模式来实现的。

什么是观察者模式？

其实就是发布订阅模式，发布者发布信息，订阅者接受信息。没订阅就接收不到信息。

观察者模式面向的需求是:A对象(观察者)对B对象(被观察者)的某种变化高度敏感，需要在B对象变化的一瞬间做出变化。程序的观察者模式，观察者不需要时刻盯着被观察者，而是采用注册(Regsiter)或者订阅(Subscribe)的方式，告诉被观察者，我是你的某种状态，你的状态在发生变化的时候来通知我。

例如：连载小说和读者的关系，观察者是读者，被观察者是连载小说，读者订阅了连载小说，当连载小说发布新的章节时通知并推送给读者，然后读者阅读新的章节，这就是观察者模式。

RxJava的观察者模式

RxJava有四个基本概念：Observer(观察者)，Observable(被观察者)，subscribe(订阅)，事件。Observer和Observable通过subscribe()实现订阅关系，从而Observable可以在需要的时候发出事件通知Observer。

  • Observer：     观察者，它决定事件发生时有怎么样的行为；
  • Observable： 被观察者，它决定什么时候出发事件以及触发什么样的事件；
  • subscribe：    订阅，将Observer和Observable关联起来。

我们来看看上面的例子的具体步骤：

1）创建被观察者Observable

它决定什么时候出发事件以及触发怎么样的事件，通过Observable.create(ObservableOnSubscribe)创建被观察者实例，这是最基本的创造事件序列的方法，并为它定义事件出发规则。

  1.  //创建被观察者Observable
  2.  Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
  3.  @Override
  4.  public void subscribe(ObservableEmitter<String> e) throws Exception {
  5.  e.onNext("RxJava：e.onNext== 第一次");
  6.  e.onNext("RxJava：e.onNext== 第二次");
  7.  e.onNext("RxJava：e.onNext== 第三次");
  8.  e.onComplete();
  9.  }
  10.  });

创建一个ObservableOnSubscribe对象并且实现subscribe()方法，设定了事件的内容是String，并返回ObservableEmitter，相当于一个计划表，当Observable被订阅的时候，复写subscribe()方法定义发送的事件，ObservableEmitter是事件发射器，定义并且观向察者发送需要发送的事件，onNext()会被执行三次，最后执行onComplete()方法，这样由被观察者调用观察者回调的方法，实现了被观察者向观察者传递事件。

2）创建观察者Observer

它决定事件触发有怎样的行为，定义响应事件的行为，直接new一个Observer观察者实例，实现其中相应的方法：

  1.  //创建观察者Observer
  2.  Observer<String> observer = new Observer<String>() {
  3.  @Override
  4.  public void onSubscribe(Disposable d) {
  5.  Log.e(TAG, "onSubscribe == 订阅");
  6.  }
  7.  
  8.  @Override
  9.  public void onNext(String s) {
  10.  Log.e(TAG, "onNext == " + s);
  11.  }
  12.  
  13.  @Override
  14.  public void onError(Throwable e) {
  15.  Log.e(TAG, "onError == " + e.getMessage());
  16.  }
  17.  
  18.  @Override
  19.  public void onComplete() {
  20.  Log.e(TAG, "onComplete == ");
  21.  }
  22.  };
  • onSubscribe(Disposable d)：  事件订阅成功回调，返回Disposable请求实例，可以通过d.dispose()取消请求；
  • onNext(T)：                             响应事件的方法，发送事件时，观察者会回调onNext()方法，接收事件数据；
  • onError()：                               事件队列异常，在处理事件出现异常的时候回触发这个方法，其他事件不会再继续发出；
  • onComplete()：                       事件队列完结，当不再有新onNext()发出时，需要触发onComplete()方法来作为标志，其他事件不会再继续发出

在一个正确的事件运行队列中，onError()和onComplete()有且仅有一个出现，还是在事件的最后出现，即onError()和onComplete()是互斥的，当一个出现了，另一个就不再回出现。

3）subscribe()

订阅，连接Observable(被观察者)和Observer(观察者)。

  1.  //订阅(观察者观察被观察者)
  2.  observable.subscribe(observer);

Observable是被观察者，observer是观察者，创建完Observer和Observable之后，通过subscribe()将两者关联起来。

通过该调用，回调观察者的相关方法，从而响应被观察者响应的事件，Observable只生产事件，真正发送事件的是在它订阅的时候，即subscribe()被调用的时候。

另外：观察者Obaserver的subscribe具有多个重载的方法：

  1.  //观察者不对被观察者发送的事件做出响应(但是被观察者还可以继续发送事件)
  2.  public final Disposable subscribe()
  3.  
  4.  //观察者对被观察者发送的任何事件都做出响应
  5.  public final void subscribe(Observer<? super T> observer)
  6.  
  7.  //表示观察者只对被观察者发送的Next事件做出响应
  8.  public final Disposable subscribe(Consumer<? super T> onNext)
  9.  
  10.  //表示观察者只对被观察者发送的Next & Error事件做出响应
  11.  public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)
  12.  
  13.  //表示观察者只对被观察者发送的Next & Error & Complete事件做出响应
  14.  public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
  15.  Action onComplete)
  16.  
  17.  //表示观察者只对被观察者发送的Next & Error & Complete & onSubscribe事件做出响应
  18.  public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
  19.  Action onComplete, Consumer<? super Disposable> onSubscribe)

2.Scheduler调度者

在RxJava默认规则中，事件的发出和消费都是在同一个线程中发生的，那么上面的例子来说，就是一个同步的观察者模式。观察者模式的本省就是后台处理，前台回调的异步机制，因此异步对RxJava来说是至关重要的，异步的实现则需要用到Scheduler调度器来切换线程。

在RxJava中Scheduler(调度器)相当于线程控制器，RxJava通过Scheduler来指定那一部分代码执行在哪一个线程。我们来看看简单的例子：

  1.  Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
  2.  @Override
  3.  public void subscribe(ObservableEmitter<String> e) throws Exception {
  4.  e.onNext("RxJava：e.onNext== 第一次"); 
  5.  e.onComplete();
  6.  Log.d(TAG, "subscribe()线程==" + Thread.currentThread().getId());
  7.  }
  8.  }).subscribeOn(Schedulers.io())//指定被观察者subscribe()(发送事件的线程)在IO线程()
  9.  .observeOn(AndroidSchedulers.mainThread());//指定观察者接收响应事件的线程在主线程

log如下：

可以看到主线程id为1，执行事件subscribe()方法的线程id为432，回调事件方法onNext()的线程为1，即在主线程中执行。

由于subscribeOn(Schedulers.io())指定了subscribe()方法发送事件线程在IO线程中执行，observeOn(AndroidSchedulers.mainThread())指定了接收事件在主线程中执行。这种方式非常常见，适用于后台获取数据，前台显示的程序策略。

  • subscribeOn()：      用于指定Observable被观察者subscribe()时所发生的线程，即指定发生事件的线程
  • observeOn()：         指定Observer观察者接收&响应事件的线程，即订阅者接收事件的线程

注意：多次指定发射事件的线程只有第一次指定有效，也就是说多次调用subscribeOn()只有第一次有效，其余的会被忽略；但是多次指定订阅者接收事件的线程是可以的，也就是说每observeOn()一次，接收事件的线程就会切换一次。

RxJava中内置了很多线程项供我们选择：

  • Schedulers.io()：        代表IO操作的线程，通常用于网络、读写文件等IO密集型的操作。行为模式和new Thread()差不多，只是IO的内部是一个无上限的线程池，可重用空闲的线程，更高效（不要把计算工作放在IO内，可以避免创建不必要的线程）；
  • AndroidSchedulers.mainThread()：Android的主线程；用于更新UI
  • Schedulers.newThread()：             总是启用新线程，并在新线程中执行操作；多用于耗时操作
  • Schedulers.computation()：           代表CPU计算密集型的操作，即不会被IO等操作限制性能的操作。

三、常用操作方式

RxJava的强大之处，在于它提供类丰富且强悍的操作符，通过使用和组合操作符，你几乎能完成你想完成的任务。下面我们来了解这些操作符的含义和使用方法。

1.快速创建

（1）create()：是所有创建型操作符的“根”，也就是说其他创建型操作符最后都是通过create()来创建的Observable的，快速创建被观察者对象，仅发送onComplete()事件，直接通知完成。

  1. Observable.create(new ObservableOnSubscribe<String>() {
  2.  @Override
  3.  public void subscribe(ObservableEmitter<String> emitter) throws Exception {
  4.  try{
  5.  if (!emitter.isDisposed()){
  6.  emitter.onNext("RxJava：e.onNext== 第一次");
  7.  emitter.onNext("RxJava：e.onNext== 第二次");
  8.  emitter.onNext("RxJava：e.onNext== 第三次");
  9.  emitter.onComplete();
  10.  }
  11.  }catch (Exception e){
  12.  emitter.onError(e);
  13.  }
  14.  }
  15.  }).subscribe(new Observer<String>() {
  16.  //默认最先复写onSubscribe()
  17.  @Override
  18.  public void onSubscribe(Disposable d) {
  19.  Log.e(TAG, "onSubscribe == 订阅");
  20.  }
  21.  
  22.  @Override
  23.  public void onNext(String s) {
  24.  Log.e(TAG, "onNext == " + s);
  25.  }
  26.  
  27.  @Override
  28.  public void onError(Throwable e) {
  29.  Log.e(TAG, "onError == " + e.getMessage());
  30.  }
  31.  
  32.  @Override
  33.  public void onComplete() {
  34.  Log.e(TAG, "onComplete == ");
  35.  }
  36.  });

在使用create()操作符时，最好在被观察者的回调函数subscribe()中加上isDisposed()，以便在观察者断开连接的时候不在执行subscribe()函数中的相关逻辑，避免意想不到的错误出现。

打印数据如下：

（2）empty()：快速创建被观察者对象，仅发送onComplete()事件，直接通知完成。

  1. //快速创建被观察者对象，仅发送onComplete()事件，直接通知完成。
  2. Observable.empty()
  3.  .subscribe(new Observer<Object>() {
  4.  @Override
  5.  public void onSubscribe(Disposable d) {
  6.  Log.e(TAG, "empty：onSubscribe == 订阅");
  7.  }
  8.  
  9.  @Override
  10.  public void onNext(Object o) {
  11.  Log.e(TAG, "empty：onNext ==" + o.toString());
  12.  }
  13.  
  14.  @Override
  15.  public void onError(Throwable e) {
  16.  Log.e(TAG, "empty：onError == " + e.getMessage());
  17.  }
  18.  
  19.  @Override
  20.  public void onComplete() {
  21.  Log.e(TAG, "empty：onComplete == ");
  22.  }
  23.  });

打印log如下：

（3）error()：快速创建被观察者对象，仅发送onError()事件，直接通知异常。

  1.  //快速创建被观察者对象，仅发送onError()事件，直接通知异常。
  2.  Observable.error(new Throwable("只回调error"))
  3.  .subscribe(new Observer<Object>() {
  4.  @Override
  5.  public void onSubscribe(Disposable d) {
  6.  Log.e(TAG, "error：onSubscribe == 订阅");
  7.  }
  8.  
  9.  @Override
  10.  public void onNext(Object o) {
  11.  Log.e(TAG, "error：onNext ==" + o.toString());
  12.  }
  13.  
  14.  @Override
  15.  public void onError(Throwable e) {
  16.  Log.e(TAG, "error：onError == " + e.getMessage());
  17.  }
  18.  
  19.  @Override
  20.  public void onComplete() {
  21.  Log.e(TAG, "error：onComplete == ");
  22.  }
  23.  });

打印log如下：

（4）never()：快速创建被观察者对象，不发送任何事件。

  1.  //快速创建被观察者对象，不发送任何事件。
  2.  Observable.never()
  3.  .subscribe(new Observer<Object>() {
  4.  @Override
  5.  public void onSubscribe(Disposable d) {
  6.  Log.e(TAG, "never：onSubscribe == 订阅");
  7.  }
  8.  
  9.  @Override
  10.  public void onNext(Object o) {
  11. 以上是关于rxjava的主要内容，如果未能解决你的问题，请参考以下文章
     

     使用 rxJava 和改造多次调用另一个请求中的请求
    

     导航到另一个片段时触发 API 调用
    

     如何在基础活动暂停时暂停rxjava Observable.interval
    

     如何在Activity中使用Retrofit和RxJava / RxAndroid处理旋转？
    

     从 recyclerview 的适配器访问父片段方法
    

     一起来造一个RxJava，揭秘RxJava的实现原理