Carson带你学Android：面向初学者的RxJava使用指南

前言

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

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

今天，Carson将为初学者带来 RxJava的基本使用，希望大家会喜欢。

Carson带你学RxJava系列文章，包括 原理、操作符、应用场景、背压等等，请关注看文章：Android：这是一份全面 & 详细的RxJava学习指南

1. RxJava简介

此处简单介绍RxJava

示意图

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

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：

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

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

3. 实例讲解

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

3.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 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事件作出响应");
            }

        });
    }
}

测试效果
实现效果同上

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

喜欢的麻烦点个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. 总结

本文主要讲解 Rxjava 的基础使用，更多实例应用请看文章：Android RxJava实际应用教学：你该什么时候使用RxJava？
Carson带你学RxJava系列文章：

入门
Carson带你学Android：这是一篇清晰易懂的Rxjava入门教程
 Carson带你学Android：面向初学者的RxJava使用指南
 Carson带你学Android：RxJava2.0到底更新了什么？
原理
Carson带你学Android：图文解析RxJava原理
 Carson带你学Android：手把手带你源码分析RxJava
使用教程：操作符
Carson带你学Android：RxJava操作符教程
 Carson带你学Android：RxJava创建操作符
 Carson带你学Android：RxJava功能性操作符
 Carson带你学Android：RxJava过滤操作符
 Carson带你学Android：RxJava组合/合并操作符
 Carson带你学Android：RxJava变换操作符
 Carson带你学Android：RxJava条件/布尔操作符
实战
Carson带你学Android：什么时候应该使用Rxjava？(开发场景汇总)
Carson带你学Android：RxJava线程控制(含实例讲解)
Carson带你学Android：图文详解RxJava背压策略
 Carson带你学Android：RxJava、Retrofit联合使用汇总(含实例教程)
Carson带你学Android：优雅实现网络请求嵌套回调
 Carson带你学Android：网络请求轮询(有条件)
Carson带你学Android：网络请求轮询(无条件)
Carson带你学Android：网络请求出错重连(结合Retrofit)
Carson带你学Android：合并数据源
 Carson带你学Android：联想搜索优化
 Carson带你学Android：功能防抖
 Carson带你学Android：从磁盘/内存缓存中获取缓存数据
 Carson带你学Android：联合判断

欢迎关注Carson_Ho的简书

不定期分享关于安卓开发的干货，追求短、平、快，但却不缺深度。

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

最后编辑于：2021.09.17 22:08:08

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 206,839评论 6赞 482
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 88,543评论 2赞 382
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 153,116评论 0赞 344
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 55,371评论 1赞 279
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 64,384评论 5赞 374
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 49,111评论 1赞 285
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,416评论 3赞 400
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 37,053评论 0赞 259
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 43,558评论 1赞 300
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 36,007评论 2赞 325
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,117评论 1赞 334
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,756评论 4赞 324
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,324评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,315评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,539评论 1赞 262
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,578评论 2赞 355
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,877评论 2赞 345

Carson带你学Android：面向初学者的RxJava使用指南

Carson带你学Android：面向初学者的RxJava使用指南

前言

目录

1. RxJava简介

2. 基本使用

2.1 方式1：分步骤实现

2.1.1 使用步骤

2.1.2 步骤详解

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

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

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

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

特别注意

3. 实例讲解

3.1 方式1：分步骤实现

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

4. 额外说明

4.1 观察者 Observer的subscribe()具备多个重载的方法

4.2 可采用 Disposable.dispose() 切断观察者与被观察者之间的连接

5. 总结

欢迎关注Carson_Ho的简书

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

推荐阅读更多精彩内容

Carson带你学Android：面向初学者的RxJava使用指南

前言

目录

1. RxJava简介

2. 基本使用

2.1 方式1：分步骤实现

2.1.1 使用步骤

2.1.2 步骤详解

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

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

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

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

特别注意

3. 实例讲解

3.1 方式1：分步骤实现

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

4. 额外说明

4.1 观察者 Observer的subscribe()具备多个重载的方法

4.2 可采用 Disposable.dispose() 切断观察者 与 被观察者 之间的连接

5. 总结

欢迎关注Carson_Ho的简书

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

推荐阅读更多精彩内容

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

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

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

4.2 可采用 Disposable.dispose() 切断观察者与被观察者之间的连接