RxJava操做符系列一

前言

第一次接触学习RxJava应该是一两个月前的事情了,但其中也是断断续续,最近又再次去学习RxJava,和当初刚接触RxJava彻底不是一样的心情,轻松了不少,也感觉到了RxJava的魅力,真是不禁衷感叹太牛了。目前关于RxJava的文章也不少,我的推荐两篇扔物线的给 Android 开发者的 RxJava 详解和大头鬼Bruce的译文 深刻浅出RxJava系列。那么这篇文章经过代码介绍RxJava中的操做符,以及操做符的使用。固然操做符较多,准备分几篇文章介绍。若是你想提早学习其余操做符能够去GitHub,欢迎star项目。javascript


RxJava操做符源码传送门


基础介绍

为了力求有没有RxJava基础都能看懂此文,简单介绍一下RxJava以及一些名词。在RxJava开源的Github上是这样解释的a library for composing asynchronous and event-based programs using observable sequences for the Java VM。不管多么复杂的逻辑,均可以保持整洁的代码格式。java

在RxJava中最重要的就是Observable(被观察者),subscribe(订阅),Observer(观察者)或者Subscriber(订阅者),Observable也就是数据(事件)源,Subscriber负责接收以及处理数据(事件)。固然要想实现二者通讯,须要有一种机制那就是订阅。Observer 经过 subscribe() 方法实现订阅关系,从而 Observable 能够在须要的时候发出事件来通知 Observer。git

例如张三(观察者)想看某款新闻软件的科技信息(被观察者),因为科技信息是天天推送或者不定时推送,若是张三一直盯着手机屏幕看而且刷新消息是否是又新的信息,显然不现实。这时候就能够经过张三 subscribe(订阅)科技信息,而实现当有新的科技信息时自动给张三推送消息,在这期间,张三并不须要一直盯着屏幕刷新闻。在咱们平时的认知中实现订阅应该是张三.subscribe(科技新闻),不过在RxJava代码中实现订阅应该写成科技新闻.subscribe(张三)。github

在RxJava中,有三个事件回调方法,分别是onNext(),OnError(),onCompleted(),onNext()是最终输出及处理数据的回调,在发射数据过程当中出现错误异常会回调OnError()方法,当不会再有新的 onNext() 发出时,须要触发 onCompleted() 方法做为标志。,OnError()和onCompleted()是互斥的。下面举一个最简单的例子数组

Observable observable2 = Observable.just("也许当初忙着微笑和哭泣", "忙着追逐天空中的流星", "人理所固然的忘记", "是谁风里雨里一直默默守护在原地");
        Subscriber subscriber = new Subscriber<String>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted: " )
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: ")
            }

            @Override
            public void onNext(String s) {
               Log.e(TAG, "onNext: "+s )
            }
        };
        observable.subscribe(subscriber);复制代码

运行后打印信息为app

onNext: 也许当初忙着微笑和哭泣
onNext: 忙着追逐天空中的流星
onNext: 人理所固然的忘记
onNext: 是谁风里雨里一直默默守护在原地
onCompleted:复制代码

Create

咱们可使用该操做符从零开始建立一个Observable,给这个操做符传递一个接受观察者做为参数的函数,并调用观察者的onNext,onError和onCompleted方法。以下async

//被观察者
        Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                //能够屡次调用subscriber.onNext("你们好")发射数据
                subscriber.onNext("你们好");
                subscriber.onNext("我开始学习RxJava");
                subscriber.onCompleted();

            }
        });复制代码

发送数据须要在毁掉方法call中调用subscriber的onNext(),onNext(T)发送的参数须要和Observable.OnSubscribe()中参数相同,在上面咱们传入的是String类型。建立后Observale后,咱们须要建立Subscriber(观察者)去处理observable发送的数据。以下ide

Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted");
            }
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, e.getMessage());
            }
            @Override
            public void onNext(String s) {
                Log.e(TAG, "onNext:"+s);
            }
        };复制代码

数据成功发送后,会回调Subscriber的onNext()的方法,其中的参数就是接收到的数据。当onNext()接收数据完毕后会执行onCompleted(),若是中途有环节出现错误异常,会执行onError()。如今观察者和被观察者都建立完毕了,他们执行还须要一个前提就是订阅,若是不订阅,observable并不会发射数据,subscribe也不会接收数据,订阅代码以下函数

observable.subscribe(subscriber);复制代码

执行后输出信息post

onNext:你们好
onNext:我开始学习RxJava
onCompleted复制代码

from

该操做符是将其它种类的对象和数据类型转换为Observable,若是当你发射的的数据是同一种类型,而不是混合使用Observables和其它类型的数据,会很是方便。以下建立Observable

Integer[] integers = {1,2, 3, 4};
        Observable<Integer> observable=Observable.from(integers);
         Subscriber<String> subscriber = new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted");
            }
            @Override
            public void onError(Throwable e) {
                Log.e(TAG, e.getMessage());
            }
            @Override
            public void onNext(Integer i) {
                Log.e(TAG, "onNext:"+i);
            }
        };
        observable.subscribe(subscriber);复制代码

输出信息为

onNext:1
onNext:2
onNext:3
onNext:4
onCompleted复制代码

from操做符能够转换Future、Iterable和数组。对于Iterable和数组,产生的Observable会发射Iterable或数组的每一项数据。对于Future,它会发射Future.get()方法返回的单个数据,而且还能够增长经过: from(Future,timeout, timeUnit)指定超时时间,若是执行的时候Future超时会回调onError()方法。

just

just将单个数据转换为发射那个数据的Observable,Just相似于From,可是From会将数组或Iterable的数据取出而后逐个发射,而Just只是简单的原样发射,将数组或Iterable当作单个数据,若是你传递null给Just,它会返回一个发射null值的Observable。不要误认为它会返回一个空Observable(彻底不发射任何数据的Observable)。对于just能够接收1到10个数据,返回一个按参数列表顺序发射这些数据的Observable。

Observable.just(1 2, 3, 4)
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.e(TAG, "onCompleted: ");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: ");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e(TAG, "onNext: " + integer);
                    }
                });复制代码

输出

onNext:1
onNext:2
onNext:3
onNext:4
onCompleted:复制代码

对于just参数类型能够是多种,以下,传入两个类型数据

Observable.just(0, "one", 6, "two", 8, "three")
                .subscribe(new Subscriber<Serializable>() {
                    @Override
                    public void onCompleted() {
                        Log.e(TAG, "onCompleted: " );
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: ");
                    }

                    @Override
                    public void onNext(Serializable serializable) {
                        Log.e(TAG, "onNext: "+serializable.toString());
                    }
                });复制代码

则输出信息

onNext:0
onNext:one
onNext:6
onNext:two
onNext:8
onNext:three
onCompleted:复制代码

Empty/Never/Error

Empty:建立一个不发射任何数据可是正常终止的Observable,此时会回调onCompleted()

Never:建立一个不发射数据也不终止的Observable

Error:建立一个不发射数据以一个错误终止的Observable

error操做符须要一个Throwable参数,你的Observable会以此终止。这些操做符默认不在任何特定的调度器上执行,可是empty和error有一个可选参数是Scheduler,若是你传递了Scheduler参数,它们会在你指定的调度器上发送通知。

Range

该操做符建立特定整数序列的Observable,它接受两个参数,一个是范围的起始值,一个是范围的数据的数目。若是你将第二个参数设为0,将致使Observable不发射任何数据(若是设置为负数,会抛异常)。

Observable.range(1,4)
                .subscribe(new Subscriber<Integer>() {
                    public String TAG="RXJAVA";

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

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: ");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e(TAG, "onNext: "+integer);
                    }
                });复制代码

输出信息

onNext: 1
onNext: 2
onNext: 3
onNext: 4
onCompleted:复制代码

你能够在代码实战中,更改第二个参数为负数,或者0,以及将第一个参数更改成你想测试的任意值,去观察执行日志帮助理解。

Timer

Timer操做符建立一个在给定的时间段以后返回一个特殊值的Observable。它在延迟一段给定的时间后发射一个简单的数字0 。

Observable.timer(1, TimeUnit.SECONDS)
                .subscribe(new Subscriber<Long>() {
                    public String TAG="RXJAVA";

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

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError: ");
                    }

                    @Override
                    public void onNext(Long integer) {
                        Log.e(TAG, "onNext:1111111 "+integer);
                    }
                });复制代码

对于该操做符默认在computation调度器上执行的,若是你想在onNext()回调方法更新UI,须要经过observeOn(AndroidSchedulers.mainThread())设置,不然会调用onError()方法。固然Time人提供的有一个三个参数的方法timer(long,TimeUnit,Scheduler)能够指定 Scheduler 。

Interval

该操做符按固定的时间间隔发射一个无限递增的整数序列,它接受一个表示时间间隔的参数和一个表示时间单位的参数,固然该操做符合Timer同样,是在computation调度器上执行的,若想更新UI须要指定Scheduler 为AndroidSchedulers.mainThread()。

Subscription   subscription = Observable.interval(1, TimeUnit.SECONDS)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<Long>() {
                    @Override
                    public void call(Long aLong) {
                        tv.append(" " + aLong + " ");
                    }
                });复制代码

经过上面代码就会每隔1秒在tv上追加一个数字,而且会永远执行。若是在某个时刻不想继续输出,就须要要解除订阅。

if (subscription != null && !subscription.isUnsubscribed()) {
            subscription.unsubscribe();
        }复制代码

Repeat

该操做符是重复的发射某个数据序列,而且能够本身设置重复的次数。当接收到onComplete()会触发重订阅再次重复发射数据,当重复发射数据次数到达后执行onCompleted()。

String[] strs = {"也许当初忙着微笑和哭泣", "忙着追逐天空中的流星"};
        Observable.from(strs).repeat(2)..subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {
                Log.e(TAG, "onCompleted: " );
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: ");
            }

            @Override
            public void onNext(String s) {
                Log.e(TAG, "onNext: "+s );
                tv1.append("\n" + s);
            }
        });复制代码

输出

onNext: 也许当初忙着微笑和哭泣
onNext: 忙着追逐天空中的流星
onNext: 也许当初忙着微笑和哭泣
onNext: 忙着追逐天空中的流星
onCompleted:复制代码

Defer

直到有观察者订阅时才建立Observable,而且为每一个观察者建立一个新的Observable,该操做符能保证订阅执行时数据源是最新的数据。以下正常代码

String test="旧数据";
        Observable observable=Observable.just(test);
        Subscriber subscriber=new Subscriber() {
            public String TAG="RXJAVA";

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

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: " );
            }

            @Override
            public void onNext(Object o) {
                Log.e(TAG, "onNext: "+o );
            }
        };
        test="新数据";
        observable.subscribe(subscriber);复制代码

输出

onNext: 旧数据
 onCompleted:复制代码

经过上面代码和输出日志发现,虽然在后面讲数据test更新为新数据,可是并无生效,要想使用最新的数据就须要使用defer操做符。此时更改使用defer

test="旧数据";
        Observable<String> observable=Observable.defer(new Func0<Observable<String>>() {
            @Override
            public Observable<String> call() {
                return Observable.just(test);
            }
        });
        Subscriber subscriber=new Subscriber() {
            public String TAG="RXJAVA";

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

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: " );
            }

            @Override
            public void onNext(Object o) {
                Log.e(TAG, "onNext: "+o );
            }
        };
        test="新数据";
        observable.subscribe(subscriber);复制代码

输出信息

onNext: 新数据
onCompleted:复制代码

经过新的打印信息,发现输出值已是最新的数据。

到这里,这篇文章暂时就先结束了,若文章有不足或者错误的地方,欢迎指正,以防止给其余读者错误引导。更多的操做符,将在接下来的几篇文章介绍。