Android Rxjava：这是一篇 清晰 & 易懂的Rxjava 入门教程

概述原文：https://www.jianshu.com/p/a406b94f3188前言Rxjava由于其基于事件流的链式调用、逻辑简洁&使用简单的特点，深受各大Android开发者的欢迎。  Github截图本文主要：面向刚接触Rxjava的初学者提供了一份清晰、简洁、易懂的Rxjava入门教程涵盖基本介绍、原 原文：https://www.jianshu.com/p/a406b94f3188前言Rxjava由于其基于事件流的链式调用、逻辑简洁 & 使用简单的特点，深受各大 AndroID开发者的欢迎。

 

 

Github截图

本文主要： 面向 刚接触Rxjava的初学者提供了一份 清晰、简洁、易懂的Rxjava入门教程

涵盖 基本介绍、原理 & 具体使用等

解决的是初学者不理解Rxjava原理 & 不懂得如何使用的问题

希望你们会喜欢。

本文主要基于Rxjava 2.0如果读者还没学习过Rxjava 1.0也没关系，因为Rxjava 2.0只是在Rxjava 1.0上增加了一些新特性，本质原理 & 使用基本相同接下来的时间，我将持续推出 AndroID中 Rxjava 2.0 的一系列文章，包括原理、 *** 作符、应用场景、背压等等 ，有兴趣可以继续关注Carson_Ho的安卓开发笔记！！

示意图

---

目录

示意图

---

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. 原理4.1 生活例子引入我用一个生活例子引入 & 讲解 Rxjava原理： 顾客到饭店吃饭

示意图

流程图

4.2 Rxjava原理介绍

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

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

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

具体原理

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

 

示意图

 

 

流程图

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

 

示意图

至此，RxJava原理讲解完毕。

---

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

5.1.1 使用步骤

 

示意图

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事件。具体如下：

 

示意图

具体实现

 

<--方式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 one rror(Throwable e) {                Log.d(TAG, "对Error事件作出响应");            }                      // 当被观察者生产Complete事件& 观察者接收到时，会调用该复写方法 进行响应            @OverrIDe            public voID onComplete() {                Log.d(TAG, "对Complete事件作出响应");            }        };<--方式2：采用Subscriber 抽象类 -->// 说明：Subscriber类 = RxJava 内置的一个实现了 Observer 的抽象类，对 Observer 接口进行了扩展// 1. 创建观察者 （Observer ）对象Subscriber<Integer> 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 one rror(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：优雅的实现方法 - 基于事件流的链式调用上述的实现方式是为了说明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 one rror(Throwable e) {                Log.d(TAG, "对Error事件作出响应");            }            @OverrIDe            public voID onComplete() {                Log.d(TAG, "对Complete事件作出响应");            }        });    }}注：整体方法调用顺序：观察者.onSubscribe（）> 被观察者.subscribe（）> 观察者.onNext（）>观察者.onComplete() 

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

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

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

特别注意

RxJava 2.x 提供了多个函数式接口 ，用于实现简便式的观察者模式。具体如下：

 

示意图

 

以 Consumer为例：实现简便式的观察者模式

 

Observable.just("hello").subscribe(new Consumer<String>() {            // 每次接收到Observable的事件都会调用Consumer.accept（）            @OverrIDe            public voID accept(String s) throws Exception {                System.out.println(s);            }        });

---

6. 实例说明

我将用一个实际工程实例来演示 Rxjava的使用

6.1 方式1：分步骤实现

步骤1：加入依赖

 

    compile 'io.reactivex.rxjava2:rxjava:2.0.1'    compile 'io.reactivex.rxjava2:rxandroID:2.0.1'

步骤2：直接在MainActivity.java中实现下述步骤

创建被观察者 （Observable ）& 生产事件创建观察者 （Observer ）并 定义响应事件的行为通过订阅（Subscribe）连接观察者和被观察者

 

public class MainActivity extends AppCompatActivity {    private static final String TAG = "Rxjava";    @OverrIDe    protected voID onCreate(Bundle savedInstanceState) {        super.onCreate(savedInstanceState);        setContentVIEw(R.layout.activity_main);// 步骤1：创建被观察者 Observable & 生产事件// 即 顾客入饭店 - 坐下餐桌 - 点菜                //  1. 创建被观察者 Observable 对象        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {            // 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();            }        });// 步骤2：创建观察者 Observer 并 定义响应事件行为// 即 开厨房 - 确定对应菜式                Observer<Integer> observer = new Observer<Integer>() {            // 通过复写对应方法来 响应 被观察者            @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 one rror(Throwable e) {                Log.d(TAG, "对Error事件作出响应");            }            @OverrIDe            public voID onComplete() {                Log.d(TAG, "对Complete事件作出响应");            }        };                // 步骤3：通过订阅（subscribe）连接观察者和被观察者        // 即 顾客找到服务员 - 点菜 - 服务员下单到厨房 - 厨房烹调        observable.subscribe(observer);

测试结果

 

示意图

6.2 方式2：基于事件流的链式调用方式

 

public class MainActivity extends AppCompatActivity {    private static final String TAG = "Rxjava";    @OverrIDe    protected voID onCreate(Bundle savedInstanceState) {        super.onCreate(savedInstanceState);        setContentVIEw(R.layout.activity_main);// 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 one rror(Throwable e) {                Log.d(TAG, "对Error事件作出响应");            }            @OverrIDe            public voID onComplete() {                Log.d(TAG, "对Complete事件作出响应");            }        });    }}

测试效果
实现效果同上

 

 

示意图

Demo 下载地址
Carson_Ho的Github地址 = RxJava2系列：基础使用

喜欢的麻烦点个star！

---

7. 额外说明7.1 被观察者 Observable的subscribe()具备多个重载的方法

 

    public final disposable subscribe() {}    // 表示观察者不对被观察者发送的事件作出任何响应（但被观察者还是可以继续发送事件）    public final disposable subscribe(Consumer<? super T> onNext) {}    // 表示观察者只对被观察者发送的Next事件作出响应    public final disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> one rror) {}     // 表示观察者只对被观察者发送的Next事件 & Error事件作出响应    public final disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> one rror, Action onComplete) {}    // 表示观察者只对被观察者发送的Next事件、Error事件 & Complete事件作出响应    public final disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> one rror, Action onComplete, Consumer<? super disposable> onSubscribe) {}    // 表示观察者只对被观察者发送的Next事件、Error事件 、Complete事件 & onSubscribe事件作出响应    public final voID subscribe(Observer<? super T> observer) {}    // 表示观察者对被观察者发送的任何事件都作出响应

7.2 可采用 disposable.dispose() 切断观察者 与 被观察者 之间的连接即观察者 无法继续 接收 被观察者的事件，但被观察者还是可以继续发送事件具体使用

 

// 主要在观察者 Observer中 实现        Observer<Integer> observer = new Observer<Integer>() {            // 1. 定义disposable类变量            private disposable mdisposable;            @OverrIDe            public voID onSubscribe(disposable d) {                Log.d(TAG, "开始采用subscribe连接");                // 2. 对disposable类变量赋值                mdisposable = d;            }            @OverrIDe            public voID onNext(Integer value) {                Log.d(TAG, "对Next事件"+ value +"作出响应"  );                if (value == 2) {                    // 设置在接收到第二个事件后切断观察者和被观察者的连接                    mdisposable.dispose();                    Log.d(TAG, "已经切断了连接：" + mdisposable.isdisposed());                }            }            @OverrIDe            public voID one rror(Throwable e) {                Log.d(TAG, "对Error事件作出响应");            }            @OverrIDe            public voID onComplete() {                Log.d(TAG, "对Complete事件作出响应");            }        };

效果图

 

示意图

---

8. 总结本文主要对 Rxjava 的入门知识进行讲解，包括基本介绍、原理 & 具体使用等接下来，我将持续推出 AndroID中 Rxjava 2.0 的一系列文章，包括原理、 *** 作符、应用场景、背压等等

 

示意图

感兴趣的同学可以继续关注本人运营的Wechat Public Account：

我想给你们介绍一个与众不同的Android微信公众号（福利回赠）我想邀请您和我一起写Android（福利回赠）

---

请点赞！因为你的鼓励是我写作的最大动力！

相关文章阅读

*** 作符使用
Android：这是一篇 清晰 & 易懂的Rxjava 入门教程
Android RxJava：最基础的 *** 作符详解 - 创建 *** 作符
Android RxJava：图文详解 变换 *** 作符
Android RxJava：组合 / 合并 *** 作符 详细教程
Android RxJava：功能性 *** 作符 全面讲解实际应用讲解
Android RxJava 实际应用讲解：（无条件）网络请求轮询
Android RxJava 实际应用讲解：（有条件）网络请求轮询
Android RxJava 实际应用讲解：网络请求嵌套回调
Android RxJava 实际应用讲解：合并数据源
Android RxJava 实际应用讲解：从磁盘 / 内存缓存中 获取缓存数据
Android RxJava 实际应用讲解：联合判断
Android RxJava：细说 线程控制（切换 / 调度 ）（含Retrofit实例讲解）
Android RxJava 实际应用讲解：网络请求出错重连（结合Retrofit）

---

  总结

以上是内存溢出为你收集整理的Android Rxjava：这是一篇 清晰 & 易懂的Rxjava 入门教程全部内容，希望文章能够帮你解决Android Rxjava：这是一篇 清晰 & 易懂的Rxjava 入门教程所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错，欢迎将内存溢出网站推荐给程序员好友。