rxjava
作者:互联网
前言:阳光总在风雨后,人生没有白吃的苦,也没有白走的路,要知道,你现在吃的苦,现在受的伤,会成为日后的徽章。
一、概述
提高开发效果,降低维护成本一直是团队追求的宗旨。现在安卓里面RxJava2+Retrofit+OKHttp的网络请求框架是最流行的,几乎一统江湖,配合响应式式编程的RxJava使用也越来越多。如果英文比较好而且有兴趣的可以到官网学习:ReactiveX的官方网站。(源码和其他相关链接在文章最后给出)
RxJava到底是什么?
RxJava在GitHub上《RxJava》的自我介绍是:a library for composing asynchronous and event-based programs using observable sequences for the Java VM.(一个在Java VM 上使用可观测的序列来组成异步的,基于事件的程序的库),有些人可能感到疑惑,其实本质上可以用一词来概括——“异步”,它就是一个异步的操作库,而别的定语都基于这之上。
RxJava的好处是啥?
同样是异步操作,为什么不用Handler或者AsyncTask等异步机制?
因为:简洁
异步操作很关键的一点是简洁性,因为在调度过程比较复杂的情况下,异步代码既难写也难读懂。Android创造的Handler和AsyncTask其实都是为了让异步代码更加简洁,RxJava的优势也是简洁,它的特别之处是:随着程序代码逻辑的变得越来越复杂,它依然保持简洁。
二、使用详解
1.基本实现
RxJava最核心的两个东西Observable(被观察者、事件源)和Observer(观察者),Observable发出一系列的事件,Observer处理这些事件。在Observer接收到事件处理之前我们很方便地对结果做出各种拦截处理等。
RxJava的流程图大致如下:
1).添加依赖库
在model的build.gradle文件内添加RxJava2依赖库,注意RxJava1与RxJava2依赖不能共存。
- implementation 'io.reactivex.rxjava2:rxjava:2.0.4'
- implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'
2).三步创建实例
- //1、创建被观察者Observable
- Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
- @Override
- public void subscribe(ObservableEmitter<String> e) throws Exception {
- e.onNext("RxJava:e.onNext== 第一次");
- e.onNext("RxJava:e.onNext== 第二次");
- e.onNext("RxJava:e.onNext== 第三次");
- e.onComplete();
- }
- });
-
- //2、创建观察者Observer
- Observer<String> observer = new Observer<String>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(String s) {
- Log.e(TAG, "onNext == " + s);
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "onComplete == ");
- }
- };
-
- //3、订阅(观察者观察被观察者)
- observable.subscribe(observer);
直接运行上面的例子,打印结果如下:
上述例子中,通过new Observer<T>()创建观察者,实现相应的方法,Observable.create(ObservableOnSubscribe)创建被观察者,subscribe()将两者关联起来。
3).观察者模式
RxJava的异步实现,是通过一种拓展的观察者模式来实现的。
什么是观察者模式?
其实就是发布订阅模式,发布者发布信息,订阅者接受信息。没订阅就接收不到信息。
观察者模式面向的需求是:A对象(观察者)对B对象(被观察者)的某种变化高度敏感,需要在B对象变化的一瞬间做出变化。程序的观察者模式,观察者不需要时刻盯着被观察者,而是采用注册(Regsiter)或者订阅(Subscribe)的方式,告诉被观察者,我是你的某种状态,你的状态在发生变化的时候来通知我。
例如:连载小说和读者的关系,观察者是读者,被观察者是连载小说,读者订阅了连载小说,当连载小说发布新的章节时通知并推送给读者,然后读者阅读新的章节,这就是观察者模式。
RxJava的观察者模式
RxJava有四个基本概念:Observer(观察者),Observable(被观察者),subscribe(订阅),事件。Observer和Observable通过subscribe()实现订阅关系,从而Observable可以在需要的时候发出事件通知Observer。
- Observer: 观察者,它决定事件发生时有怎么样的行为;
- Observable: 被观察者,它决定什么时候出发事件以及触发什么样的事件;
- subscribe: 订阅,将Observer和Observable关联起来。
我们来看看上面的例子的具体步骤:
1)创建被观察者Observable
它决定什么时候出发事件以及触发怎么样的事件,通过Observable.create(ObservableOnSubscribe)创建被观察者实例,这是最基本的创造事件序列的方法,并为它定义事件出发规则。
- //创建被观察者Observable
- Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
- @Override
- public void subscribe(ObservableEmitter<String> e) throws Exception {
- e.onNext("RxJava:e.onNext== 第一次");
- e.onNext("RxJava:e.onNext== 第二次");
- e.onNext("RxJava:e.onNext== 第三次");
- e.onComplete();
- }
- });
创建一个ObservableOnSubscribe对象并且实现subscribe()方法,设定了事件的内容是String,并返回ObservableEmitter,相当于一个计划表,当Observable被订阅的时候,复写subscribe()方法定义发送的事件,ObservableEmitter是事件发射器,定义并且观向察者发送需要发送的事件,onNext()会被执行三次,最后执行onComplete()方法,这样由被观察者调用观察者回调的方法,实现了被观察者向观察者传递事件。
2)创建观察者Observer
它决定事件触发有怎样的行为,定义响应事件的行为,直接new一个Observer观察者实例,实现其中相应的方法:
- //创建观察者Observer
- Observer<String> observer = new Observer<String>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(String s) {
- Log.e(TAG, "onNext == " + s);
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "onComplete == ");
- }
- };
- onSubscribe(Disposable d): 事件订阅成功回调,返回Disposable请求实例,可以通过d.dispose()取消请求;
- onNext(T): 响应事件的方法,发送事件时,观察者会回调onNext()方法,接收事件数据;
- onError(): 事件队列异常,在处理事件出现异常的时候回触发这个方法,其他事件不会再继续发出;
- onComplete(): 事件队列完结,当不再有新onNext()发出时,需要触发onComplete()方法来作为标志,其他事件不会再继续发出
在一个正确的事件运行队列中,onError()和onComplete()有且仅有一个出现,还是在事件的最后出现,即onError()和onComplete()是互斥的,当一个出现了,另一个就不再回出现。
3)subscribe()
订阅,连接Observable(被观察者)和Observer(观察者)。
- //订阅(观察者观察被观察者)
- observable.subscribe(observer);
Observable是被观察者,observer是观察者,创建完Observer和Observable之后,通过subscribe()将两者关联起来。
通过该调用,回调观察者的相关方法,从而响应被观察者响应的事件,Observable只生产事件,真正发送事件的是在它订阅的时候,即subscribe()被调用的时候。
另外:观察者Obaserver的subscribe具有多个重载的方法:
- //观察者不对被观察者发送的事件做出响应(但是被观察者还可以继续发送事件)
- public final Disposable subscribe()
-
- //观察者对被观察者发送的任何事件都做出响应
- public final void subscribe(Observer<? super T> observer)
-
- //表示观察者只对被观察者发送的Next事件做出响应
- public final Disposable subscribe(Consumer<? super T> onNext)
-
- //表示观察者只对被观察者发送的Next & Error事件做出响应
- public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> one rror)
-
- //表示观察者只对被观察者发送的Next & Error & Complete事件做出响应
- public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> one rror,
- Action onComplete)
-
- //表示观察者只对被观察者发送的Next & Error & Complete & onSubscribe事件做出响应
- public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> one rror,
- Action onComplete, Consumer<? super Disposable> onSubscribe)
2.Scheduler调度者
在RxJava默认规则中,事件的发出和消费都是在同一个线程中发生的,那么上面的例子来说,就是一个同步的观察者模式。观察者模式的本省就是后台处理,前台回调的异步机制,因此异步对RxJava来说是至关重要的,异步的实现则需要用到Scheduler调度器来切换线程。
在RxJava中Scheduler(调度器)相当于线程控制器,RxJava通过Scheduler来指定那一部分代码执行在哪一个线程。我们来看看简单的例子:
- Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
- @Override
- public void subscribe(ObservableEmitter<String> e) throws Exception {
- e.onNext("RxJava:e.onNext== 第一次");
- e.onComplete();
- Log.d(TAG, "subscribe()线程==" + Thread.currentThread().getId());
- }
- }).subscribeOn(Schedulers.io())//指定被观察者subscribe()(发送事件的线程)在IO线程()
- .observeOn(AndroidSchedulers.mainThread());//指定观察者接收响应事件的线程在主线程
log如下:
可以看到主线程id为1,执行事件subscribe()方法的线程id为432,回调事件方法onNext()的线程为1,即在主线程中执行。
由于subscribeOn(Schedulers.io())指定了subscribe()方法发送事件线程在IO线程中执行,observeOn(AndroidSchedulers.mainThread())指定了接收事件在主线程中执行。这种方式非常常见,适用于后台获取数据,前台显示的程序策略。
- subscribeOn(): 用于指定Observable被观察者subscribe()时所发生的线程,即指定发生事件的线程
- observeOn(): 指定
Observer观察者接收&响应事件的线程,即订阅者接收事件的线程
注意:多次指定发射事件的线程只有第一次指定有效,也就是说多次调用subscribeOn()只有第一次有效,其余的会被忽略;但是多次指定订阅者接收事件的线程是可以的,也就是说每observeOn()一次,接收事件的线程就会切换一次。
RxJava中内置了很多线程项供我们选择:
- Schedulers.io(): 代表IO操作的线程,通常用于网络、读写文件等IO密集型的操作。行为模式和new Thread()差不多,只是IO的内部是一个无上限的线程池,可重用空闲的线程,更高效(不要把计算工作放在IO内,可以避免创建不必要的线程);
- AndroidSchedulers.mainThread():Android的主线程;用于更新UI
- Schedulers.newThread(): 总是启用新线程,并在新线程中执行操作;多用于耗时操作
- Schedulers.computation(): 代表CPU计算密集型的操作,即不会被IO等操作限制性能的操作。
三、常用操作方式
RxJava的强大之处,在于它提供类丰富且强悍的操作符,通过使用和组合操作符,你几乎能完成你想完成的任务。下面我们来了解这些操作符的含义和使用方法。
1.快速创建
(1)create():是所有创建型操作符的“根”,也就是说其他创建型操作符最后都是通过create()来创建的Observable的,快速创建被观察者对象,仅发送onComplete()事件,直接通知完成。
- Observable.create(new ObservableOnSubscribe<String>() {
- @Override
- public void subscribe(ObservableEmitter<String> emitter) throws Exception {
- try{
- if (!emitter.isDisposed()){
- emitter.onNext("RxJava:e.onNext== 第一次");
- emitter.onNext("RxJava:e.onNext== 第二次");
- emitter.onNext("RxJava:e.onNext== 第三次");
- emitter.onComplete();
- }
- }catch (Exception e){
- emitter.onError(e);
- }
- }
- }).subscribe(new Observer<String>() {
- //默认最先复写onSubscribe()
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(String s) {
- Log.e(TAG, "onNext == " + s);
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "onComplete == ");
- }
- });
在使用create()操作符时,最好在被观察者的回调函数subscribe()中加上isDisposed(),以便在观察者断开连接的时候不在执行subscribe()函数中的相关逻辑,避免意想不到的错误出现。
打印数据如下:
(2)empty():快速创建被观察者对象,仅发送onComplete()事件,直接通知完成。
- //快速创建被观察者对象,仅发送onComplete()事件,直接通知完成。
- Observable.empty()
- .subscribe(new Observer<Object>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "empty:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(Object o) {
- Log.e(TAG, "empty:onNext ==" + o.toString());
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "empty:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "empty:onComplete == ");
- }
- });
打印log如下:
(3)error():快速创建被观察者对象,仅发送onError()事件,直接通知异常。
- //快速创建被观察者对象,仅发送onError()事件,直接通知异常。
- Observable.error(new Throwable("只回调error"))
- .subscribe(new Observer<Object>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "error:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(Object o) {
- Log.e(TAG, "error:onNext ==" + o.toString());
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "error:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "error:onComplete == ");
- }
- });
打印log如下:
(4)never():快速创建被观察者对象,不发送任何事件。
- //快速创建被观察者对象,不发送任何事件。
- Observable.never()
- .subscribe(new Observer<Object>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "never:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(Object o) {
- Log.e(TAG, "never:onNext ==" + o.toString());
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "never:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "never:onComplete == ");
- }
- });
打印log如下:
(5)Just(T... items):快速创建被观察者对象,最多只能发送10个事件。
- 参数类型:任意类型
- Observable.just(1, 2, 3, 4, 5).subscribe(new Observer<Integer>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "just:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(Integer integer) {
- Log.e(TAG, "just:onNext == " + integer);
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "just:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "just:onComplete == ");
- }
- });
通过just()创建传入Integer类型的参数构建Observable被观察者,相当于执行了onNext(1)~onNext(5),通过链式编程订阅观察者。log如下:
(6)fromArray(T... items):快速创建被观察者对象,可发送多个任意子类型的数据。
- 参数类型:数组,子类型可以为任意类型
- //设置需要传入的数组
- String[] strings = {"商品类","非商品类"};
- //传入数组,被观察者创建后会将数组转换成Observable并且发送里面所以的数据
- Observable.fromArray(strings).subscribe(new Observer<String>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "fromArray:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(String s) {
- Log.e(TAG, "fromArray:onNext == " + s);
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "fromArray:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "fromArray:onComplete == ");
- }
- });
通过fromArray()创建被观察者对象,传入数组,被观察者创建后会将数组转换成Observable(被观察者)并且发送里面所有的数据。打印数据如下:
(7)fromIterable(Iterable<? extends T> source):快速创建被观察者对象,可发送多个任意子类型的数据。
- 参数类型:集合,子类型可以为任意类型
- //创建集合
- List<Goods> list = new ArrayList();
- for (int i = 0; i < 3; i++) {
- Goods g = new Goods("名称" + i);
- list.add(g);
- }
- //传入集合,被观察者创建后会将数组转换成Observable并且发送里面所有的数据
- Observable.fromIterable(list).subscribe(new Observer<Goods>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "fromArray:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(Goods goods) {
- Log.e(TAG, "fromArray:onNext == " + goods.getName());
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "fromArray:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "fromArray:onComplete == ");
- }
- });
通过fromIterable()创建被观察者Observable对象,传入集合,被观察者创建后会将数组转换成Observable(被观察者)并且发送里面所有的数据。打印数据如下:
2.延迟创建
(1)defer():直到有Observer观察者订阅时,才会通过Observeable的工厂方法动态创建Observeable,并且发送事件
每次订阅后都会得到一个刚创建的最新的Observable对象,确保被观察者对象的数据是最新的。
- //1.初始化i
- Integer i = 100;
- //2.通过defer()定义被观察者(此时被观察者对象还没创建)
- Observable<Integer> defer = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
- @Override
- public ObservableSource<? extends Integer> call() throws Exception {
- return Observable.just(i);
- }
- });
-
- //3.重新设置i值
- i = 200;
-
- //4.订阅观察者(此时才会调用defer,创建被观察者对象)
- defer.subscribe(new Observer<Integer>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "defer:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(Integer i) {
- Log.e(TAG, "defer:onNext == " + i);
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "defer:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "defer:onComplete == ");
- }
- });
在订阅的时候才开始创建被观察者对象,然后再发送事件,所以i的值是最新的数据 i = 200;打印log如下:
(2)timer():快速创建Observable被观察者对象,延迟指定时间后发送一个类型为Long的事件
构造方法:
- timer(long delay, TimeUnit unit)
- timer(long delay, TimeUnit unit, Scheduler scheduler)
- delay: 延时的时间,类型为Long;
- unit: 表示时间单位,有TimeUnit.SECONDS等多种类型;
- scheduler: 表示调度器,用于指定线程。
发送的事件类型为Long,数值为0,相当于onNext(0);
- final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
-
- Log.e(TAG, "timer:当前时间 ==" + dateFormat.format(System.currentTimeMillis()));
- //延时10秒后,发送一个long值为0的事件
- Observable.timer(3, TimeUnit.SECONDS)
- .subscribe(new Observer<Long>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "timer:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(Long aLong) {
- Log.e(TAG, "timer:onNext ==" + aLong + " 时间 ==" + dateFormat.format(System.currentTimeMillis()));
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "timer:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "timer:onComplete == ");
- }
- });
延时3秒后,发送一个long值为0的事件,约3秒后响应事件onNext()接收到数据,另外,timer默认运行在一个新的线程上,也可以自定义线程调度器scheduler。打印log如下:
(3)interval():快速创建Observable被观察者对象,每隔指定的时间就发送相应的事件,事件序列从0开始,无限递增1;
构造方法:
- //在指定延迟时间后,每个多少时间发送一次事件
- interval(long initialDelay, long period, TimeUnit unit)
-
- //在指定的延迟时间后,每隔多少时间发送一次事件,可以指定调度器
- interval(long initialDelay, long period, TimeUnit unit, Scheduler scheduler)
-
- //每间隔多少时间发送一次事件,使用默认的线程
- Observable<Long> interval(long period, TimeUnit unit)
-
- //每间隔多少时间发送一次事件,可以指定调度器
- interval(long period, TimeUnit unit, Scheduler scheduler)
- initialDelay: 表示延迟开始的时间,类型为Long;
- period: 距离下一次发送事件的时间间隔,类型为Long;
- unit: 时间单位,有TimeUnit.SECONDS等多种类型;
- scheduler: 表示调度器,用于指定线程。
interval()提供了多种构造方法选择,每隔一定的时间period就发送一次事件,事件数据从0开始,无限增加1。
- //initialDelay:表示延迟开始的时间, period:距离下一次发送事件的时间间隔, unit:时间单位
- Observable.interval(3, 1, TimeUnit.SECONDS)
- .subscribe(new Observer<Long>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "interval:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(Long aLong) {
- Log.e(TAG, "interval:onNext ==" + aLong);
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "interval:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "interval:onComplete == ");
- }
- });
上述列子延迟3秒后开始发送事件,每次发送事件的时间间隔为1秒,无限发送,时间单位设置为秒,这种情景我们可以用来设置定时器等相关操作,打印log如下:
(4)intervalRange():类似于interval(),快速创建一个被观察者对象,指定时间间隔就发送事件,可以执行发送事件的数量,数据依次递增1。
构造方法:
- intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit)
-
- intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit, Scheduler scheduler)
- start: 表示事件开始的数值大小,类型为Long;
- count: 表示事件执行的次数,类型为long,不能为负数;
- initialDelay: 表示延迟开始的时间,类型为Long;
- period: 距离下一次发送事件的时间间隔,类型为Long;
- unit: 时间单位,有TimeUnit.SECONDS等多种类型;
- scheduler: 表示调度器,用于指定线程。
注意:count不能为负数,否则会抛出异常,我将count设置为-1,运行报错如下:
- //事件开始大小为10,发送5次事件,延迟3秒后执行,每次执行的间隔为1,单位为秒
- Observable.intervalRange(10, 5, 3, 1, TimeUnit.SECONDS)
- .subscribe(new Observer<Long>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "intervalRange:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(Long aLong) {
- Log.e(TAG, "intervalRange:onNext ==" + aLong);
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "intervalRange:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "intervalRange:onComplete == ");
- }
- });
这里指定了事件开始大小为10,发送5次事件,延迟3秒后执行,每次执行的间隔为1,单位为秒,打印log如下:
(5)range(final int start, final int count):类似于intervalRange(),快速创建一个被观察者对象,指定事件起始值,执行发送事件的数量,但是区别在于range()不能延迟发送的时间。
注意:上述参数的类型是int类型,rangeLong(long start, long count)只是参数类型不同,用法一致。
- //start:事件的开始值大小,count:发送的事件次数
- Observable.range(5, 4).subscribe(new Observer<Integer>() {
- @Override
- public void onSubscribe(Disposable d) {
- Log.e(TAG, "range:onSubscribe == 订阅");
- }
-
- @Override
- public void onNext(Integer integer) {
- Log.e(TAG, "range:onNext ==" + integer);
- }
-
- @Override
- public void one rror(Throwable e) {
- Log.e(TAG, "range:onError == " + e.getMessage());
- }
-
- @Override
- public void onComplete() {
- Log.e(TAG, "range:onComplete == ");
- }
- });
这里指定了事件的开始值大小为5,发送的事件数为4次,一次递增1;打印log如下:
上述的几种常用的创建操作符总结一下:
创建类型 | 作用 | 备注 | 使用场景 | |
---|---|---|---|---|
基本创建 | create() | 创建一个完成的被观察者(Observable) | RxJava中创建被观察者最基本的操作符 | 1、完整&快速创建被观察者 2、数组、集合遍历 |
快速创建 | empty() | 快速创建后只发送complete事件,直接通知完成 | ||
error() | 快速创建后只发送error事件,直接通知异常 | |||
never() | 快速创建后不发送任何事件 | |||
just() | 快速创建后直接发送传入的事件 | 参数最多只能10个,即发送的事件最多10个 | ||
fromArray() | 快速创建后直接发送传入的数组数据 | 参数为数组,子类型为任意类型,可10个以上 | ||
fromIterable() | 快速创建后直接发送传入的集合List数据 | 参数为集合List,子类型为任意类型,可10个以上 | ||
延时创建 | defer() | 直到观察者Observer订阅被观察者Observable时,才动态创建被观察者&发送事件 | 通过Observable工厂方法创建被观察者,每次订阅后都会得到一个最新创建的被观察者Observable,确保里面的数据是最新的 | 1、定时操作 2、周期性操作 |
timer() | 快速创建被观察者,指定延时时间,发送一个数值为0的事件 | 延时指定时间后发送一个参数为0的事件,相当于onNext(0) | ||
interval() | 快速创建被观察者,每隔指定时间发送事件 | 发送事件序列,从0开始,无限递增1 | ||
intervalRange() | 快速创建被观察者,每隔指定时间发送事件,可指定发送事件数 | 发送无限递增1的事件序列,可指定起始值大小和事件次数,可延时,类似interval() | ||
range() | 快速创建被观察者,连续发送一个时间序列,可指定范围 | 发送无限递增1的事件序列,可指定起始值大小和事件次数,不可延时,类似intervalRange() | ||
rangeLong() | 同上,区别在于数据参数为类型Long | 同上 |
本来想一篇写完的,但是考虑到篇幅太长影响阅读的效率,所以还是另外再开一篇来讲解其他的用法,想继续了解的可以点击RxJava2最全面、最详细的用法讲解(二)。
至此,本文结束!
源码地址:https://github.com/FollowExcellence/Rxjava_Retrofit
请尊重原创者版权,转载请标明出处:https://blog.csdn.net/m0_37796683/article/details/102525484 谢谢!
相关文章:
- Retrofit2的介绍和简单使用
- OKHttp3的用法介绍和解析
- 从源码角度解释OKHttp3的关键流程和重要操作
- 详细介绍了RxJava的使用(基本创建、快速创建、延迟创建等操作符)
- RxJava转换、组合、合并等操作符的使用
- RxJava延迟、do相关、错误处理等操作符的使用
- RxJava过滤、其他操作符的使用
上述几篇都是android开发必须掌握的,后续会完善其他部分!
标签:onNext,观察者,public,Override,TAG,rxjava,void 来源: https://www.cnblogs.com/jobim/p/14843553.html