RxJava——入门篇

一、ReactiveX简介

在学习RxJava前首先需要了解ReactiveX,因为RxJava是ReactiveX的一种Java的实现形式

  • ReactiveX的官网地址为:
 http://reactivex.io/

ReactiveX官网对于自身的介绍是:

An API for asynchronous programming with observable streams

实质上我们可以对其解读为三部分:

ReactiveX的解读
①An API: 首先它是个编程接口,不同语言提供不同实现。例如Java中的RxJava。
②For asynchronous programming: 在异步编程设计中使用。例如开启子线程处理耗时的网络请求。
③With observable streams: 基于可观察的事件流。例如观察者模式中观察者对被观察者的监听。

而ReactiveX结合了如下三部分内容:

  1. 观察者模式,即定义对象间一种一对多的依赖关系,当一个对象改变状态时,则所有依赖它的对象都会被改变。
  2. Iterator模式,即迭代流式编程模式。
  3. 函数式编程模式,即提供一系列函数样式的方法供快速开发。

Reactive的模式图如下:

图1.1 ReactiveX的模式图

二、RxJava的使用

1、RxJava的优势

在Android的SDK中,给开发者提供的用于异步操作的原生内容有AsyncTask和Handler。对于简单的异步请求来说,使用Android原生的AsyncTask和Handler即可满足需求,但是对于复杂的业务逻辑而言,依然使用AsyncTask和Handler会导致代码结构混乱,代码的可读性非常差。
但是RxJava的异步操作是基于观察者模式实现的,在越来越复杂的业务逻辑中,RxJava依旧可以保持简洁

2、RxJava的配置

首先,在Android Studio中配置Module的build.gradle,在这里我们使用的版本是1.2版本,并且导入RxAndroid,辅助RxJava完成线程调度:

        implementation "io.reactivex:rxjava:1.2.0"
        implementation "io.reactivex:rxandroid:1.2.0"

然后,RxJava基于观察者设计模式,其中的关键性三个步骤如下:

(1)Observable被观察者

Observable被观察者创建的代码如下:

        Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("Alex");
                subscriber.onCompleted();
            }
        });

在这里,要强调的是Observable被观察者是类类型,其中有诸多方法,我们关注其构造函数与创建Observable对象的方法,查看如下图对应的视图结构:
图2.2.1 Observable被观察者对象的视图结构

查看源码:

        protected Observable(OnSubscribe<T> f) {
           this.onSubscribe = f;
        }

        public interface OnSubscribe<T> extends Action1<Subscriber<? super T>> {
        
        }
        public static <T> Observable<T> create(OnSubscribe<T> f) {
           return new Observable<T>(RxJavaHooks.onCreate(f));
        }
  
        public static <S, T> Observable<T> create(SyncOnSubscribe<S, T> syncOnSubscribe) {
           return create((OnSubscribe<T>)syncOnSubscribe);
        }

        public static <S, T> Observable<T> create(AsyncOnSubscribe<S, T> asyncOnSubscribe) {
           return create((OnSubscribe<T>)asyncOnSubscribe);
        }

通过源码分析,可知Observable提供了create()方法来获取Observable实例对象。
此外,除了基本的创建的方法,Observable还提供了便捷的创建Observable序列的两种方式,代码如下:

  • 第一种,会将参数逐个发送
        Observable<String> observable1 = Observable.just("Alex","Payne");
  • 第二种,会将数组元素逐个转换完毕后逐个发送
        String[] observableArr = {"Alex", "Payne"};
        Observable<String> observable2 = Observable.from(observableArr);

其中Observable.just()方法会调用from()方法,详情可查看源码。

(2)Observer观察者

Observer观察者创建的代码如下:

        Observer<String> observer = new Observer<String>() {

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError,Error Info is:" + e.getMessage());
            }

            @Override
            public void onNext(String s) {
                Log.e(TAG, s);
            }
        };

Observer是接口,其中包含的方法有onCompleted()、onNext()、onError()。查看如下图所示Observer的视图结构:

图2.2.2 Observer观察者对象的视图结构

那么在RxJava中,Observer有其接口实现类对象Subscriber,它们的使用onNext、onCompleted、onError方法是一样的,但是Subscriber对于Observer接口进行了拓展,在 RxJava 的 subscribe 过程中,Observer 也总是会先被转换成一个 Subscriber 再使用,代码如下:

        Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onStart() {
                Log.e(TAG, "onStart");
            }

            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError,Error Info is:" + e.getMessage());

            }

            @Override
            public void onNext(String s) {
                Log.e(TAG, s);
            }
        };

其中,onStart()方法会在事件未发送前被调用,可以用于订阅关系建立前的准备工作,例如将数据清空或者重置,在Subscriber中默认是空实现,我们可以在该方法中调用自己的业务逻辑代码。在如下的视图结构中我们可以看到Subscriber的拓展内容,重点是add()、unsubscribe()方法以及名为subscription的Subscription队列

图2.2.3 Subscriber对象视图结构

(3)Subscribe订阅关系

Observable与observer形成订阅关系代码如下:

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

那么我们以observable.subscribe(observer)为例在这里继续查看源码,查看subscribe()方法到底做了什么:


图2.3.1 Observable调用Subscribe将Observer转换为Subscriber对象

Observer转换为Subscriber对象在这里得到印证。

  • 在之后的内容中统一以Subscriber作为订阅观察者对象

继续深入,我们可以看到订阅关系中的关键步骤(仅核心代码):

            subscriber.onStart();

            RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);

            return RxJavaHooks.onObservableReturn(subscriber);

在这里RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber)等价于OnSubscribe.call(subscriber),见下图2.3.2:

图2.3.2 RxJavaHooks.onObservableStart()转换为OnSubscribe
在return RxJavaHooks.onObservableReturn(subscriber)这里等价于return subscription,见下图2.3.3:
图2.3.3 RxJavaHooks.onObservableReturn()转换为Subscrition

  • 可以看到,subscriber() 做了3件事:
  1. 调用 Subscriber.onStart() 。该方法用于在订阅关系建立之前的准备。
  2. 调用 Observable 中的 OnSubscribe.call(Subscriber) 。OnSubscribe是Observable的内部接口,而事件发送的逻辑在这里开始运行。从这也可以看出,在 RxJava 中当 subscribe() 方法执行的时候订阅关系确立,Observable 开始发送事件。
  3. 将传入的 Subscriber 作为 Subscription 返回。这是为了方便后续调用unsubscribe()。

三、RxJava的不完整回调

1、不完整回调的代码示例

        Observable<String> observable = Observable.just("Alex", "Payne");
        Action1<String> onNextAction = new Action1<String>() {
            @Override
            public void call(String s) {
                Log.e(TAG, s);
            }
        };
        Action1<Throwable> onErrorAction = new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                Log.e(TAG, "onError,Error Info is:" + throwable.getMessage());
            }
        };
        Action0 onCompletedAction = new Action0() {
            @Override
            public void call() {
                Log.e(TAG, "onCompleted");
            }
        };

        // 根据onNextAction 来定义 onNext()
        observable.subscribe(onNextAction);
        // 根据onNextAction 来定义 onNext()、根据onErrorAction 来定义 onError()
        observable.subscribe(onNextAction, onErrorAction);
        // 根据onNextAction 来定义 onNext()、根据onErrorAction 来定义 onError()、onCompletedAction 来定义 onCompleted()
        observable.subscribe(onNextAction, onErrorAction, onCompletedAction);

2、不完整回调的原理分析

在这里我们可以看到:

Action0无参数泛型无返回值类型,而Subscriber中的onCompleted()方法也没有参数泛型
Action1有1个参数泛型无返回值类型 ,onNextAction设置的参数泛型为String,而Subscriber中的onNext()方法参数泛型也是String(和本文中观察者对象中的OnNext方法对比)
Action1有1个参数泛型无返回值类型,onErrorAction设置的参数泛型为Throwable,而Subscriber中的onError()方法参数泛型也是Throwable

那么,我们来查看observable.subscribe(onNextAction)的源码,在这里, Action1可以被当成一个包装对象,将onNext()方法进行包装作为不完整的回调传入到observable.subscribe()中

图3.2.1 传入的onNextAction最终被包装成ActionSubscriber

我们来看看Action1有何玄机,Action1的源码如下图所示:

图3.2.2 Action1接口源码
实质上,这种根据参数泛型的个数且无返回值类型的包装在RxJava中有多种如下图所示的体现,例如Action0的参数个数为0,Action1的参数个数为1以此类推:
图3.2.3 根据参数泛型的个数且无返回值类型的包装

四、RxJava的线程切换

1、Scheduler线程调度器

如果不指定线程,默认是在调用subscribe方法的线程上进行回调,那么如果子线程中调用subscibe方法,而想在主线程中进行UI更新,则会抛出异常。当然了RxJava已经帮我们考虑到了这一点,所以提供了Scheduler线程调度器帮助我们进行线程之间的切换。
实质上,Scheduler线程调度器和RxJava的操作符有紧密关联,我将在下一篇文章中进行详细介绍。

  • RxJava内置了如下所示几个的线程调度器:
  1. Schedulers.immediate():在当前线程中执行
  2. Schedulers.newThread():启动新线程,在新线程中进行操作
  3. Schedulers.io(): I/O 操作(读写文件、读写数据库、网络信息交互等)所使用的 Scheduler。行为模式和 newThread() 差不多,区别在于 io() 的内部实现是是用一个无数量上限的线程池,可以重用空闲的线程,因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中,可以避免创建不必要的线程。
  4. Schedulers.computation():计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算,即不会被 I/O 等操作限制性能的操作,例如图形的计算。这个 Scheduler 使用的固定的线程池,大小为 CPU 核数。不要把 I/O 操作放在 computation() 中,否则 I/O 操作的等待时间会浪费 CPU。
  5. Schedulers.trampoline():会将任务按添加顺序依次放入当前线程中等待执行。线程一次只执行一个任务,其余任务排队等待,一个任务都执行完成后再开始下一个任务的执行。
  • 此外RxJava还提供了用于测试的调度器Schedulers.test() 及 可自定义Scheduler—-Schedulers.form() 。

RxAndroid并且其为我们提供了AndroidSchedulers.mainThread()进行主线程的回调

2、线程控制

调用Observable对象的subscribeOn()、observeOn()方法即可完成线程控制。

  • subscribeOn(): 指定 subscribe() 所发生的线程,即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
  • observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。
            Observable.just("Alex", "Payne")
           .subscribeOn(Schedulers.io())//指定 subscribe() 所发生的线程
           .unsubscribeOn(Schedulers.io())//事件发送完毕后,及时取消发送
           .observeOn(AndroidSchedulers.mainThread())//指定 Subscriber 所运行在的线程
           .subscribe(new Action1<String>() {
               @Override
                 public void call(String s) {
                     Log.e(TAG, s);
                }
            });

五、总结

本文主要介绍了RxJava的由来、使用步骤、部分内容的原理解析。在下篇文章中我会详细介绍RxJava的操作符。希望本文对你在学习RxJava的路上有所启发。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,456评论 5 477
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,370评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,337评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,583评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,596评论 5 365
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,572评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,936评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,595评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,850评论 1 297
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,601评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,685评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,371评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,951评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,934评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,167评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 43,636评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,411评论 2 342

推荐阅读更多精彩内容

  • 前言我从去年开始使用 RxJava ,到现在一年多了。今年加入了 Flipboard 后,看到 Flipboard...
    占导zqq阅读 9,158评论 6 151
  • 我从去年开始使用 RxJava ,到现在一年多了。今年加入了 Flipboard 后,看到 Flipboard 的...
    Jason_andy阅读 5,451评论 7 62
  • 最近项目里面有用到Rxjava框架,感觉很强大的巨作,所以在网上搜了很多相关文章,发现一片文章很不错,今天把这篇文...
    Scus阅读 6,853评论 2 50
  • 文章转自:http://gank.io/post/560e15be2dca930e00da1083作者:扔物线在正...
    xpengb阅读 7,017评论 9 73
  • 她一直向往美好的爱情,可是即便他付出了一切,仍旧不能感化他那颗冰冷的心,直到满身伤痕累累,她才知道这段感情于她来说...
    圆润润润润润阅读 249评论 0 1