Android RxJava：这是一份RxJava使用入门学习指南

时间  2019-11-17
标签 android rxjava 这是 一份 使用 入门 学习 指南 栏目 Android 繁體版
原文   原文链接

前言

  • Rxjava因为其基于事件流的链式调用、逻辑简洁 & 使用简单的特色，深受各大 Android开发者的欢迎。

Github截图

若是还不了解RxJava，请看文章：Android：这是一篇 清晰 & 易懂的Rxjava 入门教程java

  • 今天，我将为初学者带来 RxJava的基本使用，但愿你们会喜欢。
  1. 本系列文章主要基于 Rxjava 2.0
  2. 接下来的时间，我将持续推出 Android中 Rxjava 2.0 的一系列文章，包括原理、操做符、应用场景、背压等等 ，有兴趣能够继续关注Carson_Ho的安卓开发笔记！！

示意图

目录

示意图

1. RxJava简介

此处简单介绍RxJava react

示意图

若还不了解RxJava，请看文章：Android：这是一篇 清晰 & 易懂的Rxjava 入门教程android

2. 基本使用

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

2.1 方式1：分步骤实现

2.1.1 使用步骤

示意图

2.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 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() 方法执行时
}
复制代码

2.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：git

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

更重要的是，随着程序逻辑的复杂性提升，它依然可以保持简洁 & 优雅。因此，通常建议使用这种基于事件流的链式调用方式实现RxJava。github

特别注意

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

示意图

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

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

3. 实例讲解

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

3.1 方式1：分步骤实现

步骤1：加入依赖ide

compile 'io.reactivex.rxjava2:rxjava:2.0.1'
    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
复制代码

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

  1. 建立被观察者 （Observable ）& 生产事件
  2. 建立观察者 （Observer ）并 定义响应事件的行为
  3. 经过订阅（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 onError(Throwable e) {
                Log.d(TAG, "对Error事件做出响应");
            }

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

        
        // 步骤3：经过订阅（subscribe）链接观察者和被观察者
        // 即 顾客找到服务员 - 点菜 - 服务员下单到厨房 - 厨房烹调
        observable.subscribe(observer);
复制代码
  • 测试结果

示意图

3.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 onError(Throwable e) {
                Log.d(TAG, "对Error事件做出响应");
            }

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

        });
    }
}
复制代码

喜欢的麻烦点个star！

4. 额外说明

4.1 观察者 Observer的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> onError) {} 
    // 表示观察者只对被观察者发送的Next事件 & Error事件做出响应

    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
    // 表示观察者只对被观察者发送的Next事件、Error事件 & Complete事件做出响应

    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
    // 表示观察者只对被观察者发送的Next事件、Error事件 、Complete事件 & onSubscribe事件做出响应

    public final void subscribe(Observer<? super T> observer) {}
    // 表示观察者对被观察者发送的任何事件都做出响应
复制代码

4.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 onError(Throwable e) {
                Log.d(TAG, "对Error事件做出响应");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件做出响应");
            }
        };
复制代码
  • 效果图

示意图

至此，关于RxJava的基础使用讲解完毕。

5. 总结

示意图

请帮顶 / 评论点赞！由于你的鼓励是我写做的最大动力！

欢迎关注carson_ho的微信公众号

示意图

示意图
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程