1. 定义

   RxJava 在 GitHub 的介绍：

RxJava：a library for composing asynchronous and event-based programs using observable sequences for the Java VM
// 翻译：RxJava 是一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库

总结：RxJava 是一个 基于事件流、实现异步操作的库

2. 作用

实现异步操作

 类似于 Android中的 AsyncTask 、Handler作用

3. 特点

由于 RxJava的使用方式是：基于事件流的链式调用，所以使得 RxJava：

逻辑简洁
实现优雅
使用简单

更重要的是，随着程序逻辑的复杂性提高，它依然能够保持简洁 & 优雅

4.1 生活例子引入

• 我用一个生活例子引入 & 讲解 Rxjava原理： 顾客到饭店吃饭

image.png image.png

4.2 Rxjava原理介绍

• Rxjava原理 基于 一种扩展的观察者模式

• Rxjava的扩展观察者模式中有4个角色：

角色	作用	类比
被观察者（Observable）	产生事件	顾客
观察者（Observer）	接收事件，并给出响应动作	厨房
订阅（Subscribe）	连接 被观察者 & 观察者	服务员
事件（Event）	被观察者 & 观察者 沟通的载体	菜式

• 具体原理

请结合上述 顾客到饭店吃饭 的生活例子理解：

image.png image.png

即RxJava原理可总结为：被观察者 （Observable） 通过 订阅（Subscribe） 按顺序发送事件 给观察者 （Observer）， 观察者（Observer） 按顺序接收事件 & 作出对应的响应动作。具体如下图：

image.png

至此，RxJava原理讲解完毕。

5. 基本使用

• 本文只关注 RxJava的基本使用，更深入的RxJava使用请继续关注Carson_Ho的RxJava系列
• Rxjava的使用方式有两种：
  1. 分步骤实现：该方法主要为了深入说明Rxjava的原理 & 使用，主要用于演示说明
  2. 基于事件流的链式调用：主要用于实际使用

5.1 方式1：分步骤实现

5.1.1 使用步骤

[图片上传失败...(image-3539ac-1526977136183)]

5.1.2 步骤详解

步骤1：创建被观察者 （Observable ）& 生产事件

• 即 顾客入饭店 - 坐下餐桌 - 点菜
• 具体实现

        // 1\. 创建被观察者 Observable 对象
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
          // create() 是 RxJava 最基本的创造事件序列的方法
          // 此处传入了一个 OnSubscribe 对象参数
          // 当 Observable 被订阅时，OnSubscribe 的 call() 方法会自动被调用，即事件序列就会依照设定依次被触发
          // 即观察者会依次调用对应事件的复写方法从而响应事件
          // 从而实现被观察者调用了观察者的回调方法 & 由被观察者向观察者的事件传递，即观察者模式

        // 2\. 在复写的subscribe（）里定义需要发送的事件
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                // 通过 ObservableEmitter类对象产生事件并通知观察者
                // ObservableEmitter类介绍
                    // a. 定义：事件发射器
                    // b. 作用：定义需要发送的事件 & 向观察者发送事件
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });

<--扩展：RxJava 提供了其他方法用于 创建被观察者对象Observable -->
// 方法1：just(T...)：直接将传入的参数依次发送出来
  Observable observable = Observable.just("A", "B", "C");
  // 将会依次调用：
  // onNext("A");
  // onNext("B");
  // onNext("C");
  // onCompleted();

// 方法2：from(T[]) / from(Iterable<? extends T>) : 将传入的数组 / Iterable 拆分成具体对象后，依次发送出来
  String[] words = {"A", "B", "C"};
  Observable observable = Observable.from(words);
  // 将会依次调用：
  // onNext("A");
  // onNext("B");
  // onNext("C");
  // onCompleted();

步骤2：创建观察者 （Observer ）并 定义响应事件的行为

• 即 开厨房 - 确定对应菜式
• 发生的事件类型包括：Next事件、Complete事件 & Error事件。具体如下：

image.png

• 具体实现

<--方式1：采用Observer 接口 -->
        // 1\. 创建观察者 （Observer ）对象
        Observer<Integer> observer = new Observer<Integer>() {
        // 2\. 创建对象时通过对应复写对应事件方法 从而 响应对应事件

            // 观察者接收事件前，默认最先调用复写 onSubscribe（）
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "开始采用subscribe连接");
            }

            // 当被观察者生产Next事件 & 观察者接收到时，会调用该复写方法 进行响应
            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "对Next事件作出响应" + value);
            }

            // 当被观察者生产Error事件& 观察者接收到时，会调用该复写方法 进行响应
            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            // 当被观察者生产Complete事件& 观察者接收到时，会调用该复写方法 进行响应
            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        };

<--方式2：采用Subscriber 抽象类 -->
// 说明：Subscriber类 = RxJava 内置的一个实现了 Observer 的抽象类，对 Observer 接口进行了扩展

// 1\. 创建观察者 （Observer ）对象
Subscriber<String> subscriber = new Subscriber<Integer>() {

// 2\. 创建对象时通过对应复写对应事件方法 从而 响应对应事件
            // 观察者接收事件前，默认最先调用复写 onSubscribe（）
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "开始采用subscribe连接");
            }

            // 当被观察者生产Next事件 & 观察者接收到时，会调用该复写方法 进行响应
            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "对Next事件作出响应" + value);
            }

            // 当被观察者生产Error事件& 观察者接收到时，会调用该复写方法 进行响应
            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            // 当被观察者生产Complete事件& 观察者接收到时，会调用该复写方法 进行响应
            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        };

<--特别注意：2种方法的区别，即Subscriber 抽象类与Observer 接口的区别 -->
// 相同点：二者基本使用方式完全一致（实质上，在RxJava的 subscribe 过程中，Observer总是会先被转换成Subscriber再使用）
// 不同点：Subscriber抽象类对 Observer 接口进行了扩展，新增了两个方法：
    // 1\. onStart()：在还未响应事件前调用，用于做一些初始化工作
    // 2\. unsubscribe()：用于取消订阅。在该方法被调用后，观察者将不再接收 & 响应事件
    // 调用该方法前，先使用 isUnsubscribed() 判断状态，确定被观察者Observable是否还持有观察者Subscriber的引用，如果引用不能及时释放，就会出现内存泄露

步骤3：通过订阅（Subscribe）连接观察者和被观察者

• 即 顾客找到服务员 - 点菜 - 服务员下单到厨房 - 厨房烹调
• 具体实现

observable.subscribe(observer);
 // 或者 observable.subscribe(subscriber)；

• 扩展说明

<-- Observable.subscribe(Subscriber) 的内部实现 -->

public Subscription subscribe(Subscriber subscriber) {
    subscriber.onStart();
    // 步骤1中 观察者  subscriber抽象类复写的方法，用于初始化工作
    onSubscribe.call(subscriber);
    // 通过该调用，从而回调观察者中的对应方法从而响应被观察者生产的事件
    // 从而实现被观察者调用了观察者的回调方法 & 由被观察者向观察者的事件传递，即观察者模式
    // 同时也看出：Observable只是生产事件，真正的发送事件是在它被订阅的时候，即当 subscribe() 方法执行时
}

5.2 方式2：优雅的实现方法 - 基于事件流的链式调用

5.2 方式2：优雅的实现方法 - 基于事件流的链式调用

上述的实现方式是为了说明Rxjava的原理 & 使用
在实际应用中，会将上述步骤&代码连在一起，从而更加简洁、更加优雅，即所谓的 RxJava基于事件流的链式调用

// RxJava的链式操作
Observable.create(new ObservableOnSubscribe<Integer>() {
// 1. 创建被观察者 & 生产事件
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
// 2. 通过通过订阅（subscribe）连接观察者和被观察者
// 3. 创建观察者 & 定义响应事件的行为
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
// 默认最先调用复写的 onSubscribe（）

        @Override
        public void onNext(Integer value) {
            Log.d(TAG, "对Next事件"+ value +"作出响应"  );
        }

        @Override
        public void onError(Throwable e) {
            Log.d(TAG, "对Error事件作出响应");
        }

        @Override
        public void onComplete() {
            Log.d(TAG, "对Complete事件作出响应");
        }

    });
}

}

注：整体方法调用顺序：观察者.onSubscribe（）> 被观察者.subscribe（）> 观察者.onNext（）>观察者.onComplete()

这种 基于事件流的链式调用，使得RxJava：

逻辑简洁
实现优雅
使用简单

更重要的是，随着程序逻辑的复杂性提高，它依然能够保持简洁 & 优雅。所以，一般建议使用这种基于事件流的链式调用方式实现RxJava。

参考
链接：https://www.jianshu.com/p/a406b94f3188