网站上线 流程,网站左侧漂浮代码,个人主页经典句子,网站建设好后为什么要维护本篇文章已授权微信公众号 YYGeeker 独家发布转载请标明出处 CSDN学院课程地址 RxJava2从入门到精通-初级篇:edu.csdn.net/course/deta…RxJava2从入门到精通-中级篇:edu.csdn.net/course/deta…RxJava2从入门到精通-进阶篇:edu.csdn.net/course/deta…RxJava2从入门到精通-源码…本篇文章已授权微信公众号 YYGeeker 独家发布转载请标明出处 CSDN学院课程地址 RxJava2从入门到精通-初级篇:edu.csdn.net/course/deta…RxJava2从入门到精通-中级篇:edu.csdn.net/course/deta…RxJava2从入门到精通-进阶篇:edu.csdn.net/course/deta…RxJava2从入门到精通-源码分析篇:edu.csdn.net/course/deta… 3. RxJava操作符 RxJava操作符也是其精髓之一可以通过一个简单的操作符实现复杂的业务逻辑甚至还可以将操作符组合起来(即RxJava的组合过程)完成更为复杂的业务需求。比如我们前面用到的.create().subscribeOn().observeOn().subscribe()都是RxJava的操作符之一下面我们将对RxJava的操作符进行分析 掌握RxJava操作符前首先要学会看得懂RxJava的图片图片是RxJava主导的精髓下面我们通过例子说明 这张图片我们先要分清楚概念上的东西上下两行横向的直线区域代表着事件流上面一行(上游)是我们的被观察者Observable下面一行(下游)是我们的观察者Observer事件流就是从上游的被观察者发送给下游的观察者的。而中间一行的flatMap区域则是我们的操作符部分它可以对我们的数据进行变换操作。最后数据流则是图片上的圆形、方形、菱形等区域也是从上游流向下游的不同的形状代表着不同的数据类型 这张图片并不是表示没有被观察者Observable而是Create方法本身就是创建了被观察者所以可以将被观察者的上游省略。在进行事件的onNext()分发后执行onComplete()事件这样就表示事件流已经结束后续如果上游继续发事件则下游表示不接收。当事件流的onCompleted()或者onError()正好被调用过一次后此后就不能再调用观察者的任何其它回调方法 在理解RxJava操作符之前需要将这几个概念弄明白整个操作符的章节都是围绕这几个概念进行的 事件流通过发射器发射的事件从发射事件到结束事件的过程这一过程称为事件流数据流通过发射器发射的数据从数据输入到数据输出的过程这一过程称为数据流被观察者事件流的上游即Observable事件流开始的地方和数据流发射的地方观察者事件流的下游即Observer事件流结束的地方和数据流接收的地方3.1 Creating Observables (创建操作符) 1、create Observable最原始的创建方式创建出一个最简单的事件流可以使用发射器发射特定的数据类型 public static void main(String[] args) {Observable.create(new ObservableOnSubscribeInteger() {Overridepublic void subscribe(NonNull ObservableEmitterInteger e) throws Exception {for (int i 1; i 5; i) {e.onNext(i);}e.onComplete();}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}}, new ConsumerThrowable() {Overridepublic void accept(Throwable throwable) throws Exception {}}, new Action() {Overridepublic void run() throws Exception {System.out.println(onComplete);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onComplete
复制代码2、from 创建一个事件流并发出特定类型的数据流其发射的数据流类型有如下几个操作符 public static void main(String[] args) {Observable.fromArray(new Integer[]{1, 2, 3, 4, 5}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onNext5
复制代码3、just just操作符和from操作符很像只是方法的参数有所差别它可以接受多个参数 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onNext5
复制代码4、defer defer与just的区别是just是直接将发射当前的数据流而defer会等到订阅的时候才会去执行它的call()回调再去发射当前的数据流。复杂点的理解就是defer操作符是将一组数据流在原有的事件流基础上缓存一个新的事件流直到有人订阅的时候才会创建它缓存的事件流 public static void main(String[] args) {i 10;ObservableInteger just Observable.just(i, i);ObservableObject defer Observable.defer(new CallableObservableSource?() {Overridepublic ObservableSource? call() throws Exception {//缓存新的事件流return Observable.just(i, i);}});i 15;just.subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});defer.subscribe(new ConsumerObject() {Overridepublic void accept(Object o) throws Exception {System.out.println(onNext (int) o);}});i 20;defer.subscribe(new ConsumerObject() {Overridepublic void accept(Object o) throws Exception {System.out.println(onNext (int) o);}});
}
复制代码输出 onNext10
onNext10
onNext15
onNext15
onNext20
onNext20
复制代码5、interval interval操作符是按固定的时间间隔发射一个无限递增的整数数据流由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行interval默认在computation调度器上执行 public void interval() {Observable.interval(1, TimeUnit.SECONDS).subscribe(new ConsumerLong() {Overridepublic void accept(Long aLong) throws Exception {System.out.println(onNext aLong);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
......
复制代码6、range range操作符发射一个范围内的有序整数数据流你可以指定范围的起始和长度 public static void main(String[] args) {Observable.range(1, 5).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onNext5
复制代码7、repeat repeat操作符可以重复发送指定次数的某个事件流repeat操作符默认在trampoline调度器上执行 public static void main(String[] args) {Observable.just(1).repeat(5).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
onNext1
onNext1
onNext1
onNext1
复制代码8、timer timer操作符可以创建一个延时的事件流由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行默认在computation调度器上执行 public void timer() {Observable.timer(5, TimeUnit.SECONDS).subscribe(new ConsumerLong() {Overridepublic void accept(Long aLong) throws Exception {System.out.println(onNext aLong);}});
}
复制代码输出 onNext0
复制代码9、小结 create():创建最简单的事件流from():创建事件流可发送不同类型的数据流just():创建事件流可发送多个参数的数据流defer():创建事件流可缓存可激活事件流interval():创建延时重复的事件流range():创建事件流可发送范围内的数据流repeat():创建可重复次数的事件流timer():创建一次延时的事件流补充interval()、timer()、delay()的区别 interval():用于创建事件流周期性重复发送timer():用于创建事件流延时发送一次delay():用于事件流中可以延时某次事件流的发送3.2 Transforming Observables (转换操作符) 1、map map操作符可以将数据流进行类型转换 public static void main(String[] args) {Observable.just(1).map(new FunctionInteger, String() {Overridepublic String apply(Integer integer) throws Exception {return 发送过来的数据会被变成字符串 integer;}}).subscribe(new ConsumerString() {Overridepublic void accept(String s) throws Exception {System.out.println(onNext s);}});
}
复制代码输出 onNext发送过来的数据会被变成字符串1
复制代码2、flatMap flatMap操作符将数据流进行类型转换然后将新的数据流传递给新的事件流进行分发这里通过模拟请求登录的延时操作进行说明由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public void flatMap() {Observable.just(new UserParams(hensen, 123456)).flatMap(new FunctionUserParams, ObservableSourceString() {Overridepublic ObservableSourceString apply(UserParams userParams) throws Exception {return Observable.just(userParams.username 登录成功).delay(2, TimeUnit.SECONDS);}}).subscribe(new ConsumerString() {Overridepublic void accept(String s) throws Exception {System.out.println(s);}});
}public static class UserParams {public UserParams(String username, String password) {this.username username;this.password password;}public String username;public String password;
}
复制代码输出 hensen登录成功
复制代码补充 concatMap与flatMap功能一样唯一的区别就是concatMap是有序的flatMap是乱序的3、groupBy groupBy操作符可以将发射出来的数据项进行分组并将分组后的数据项保存在具有key-value映射的事件流中。groupBy具体的分组规则由groupBy操作符传递进来的函数参数Function所决定的它可以将key和value按照Function的返回值进行分组返回一个具有分组规则的事件流GroupedObservable注意这里分组出来的事件流是按照原始事件流的顺序输出的我们可以通过sorted()对数据项进行排序然后输出有序的数据流。 public static void main(String[] args) {Observable.just(java, c, c, c#, javaScript, Android).groupBy(new FunctionString, Character() {Overridepublic Character apply(String s) throws Exception {return s.charAt(0);//按首字母分组}}).subscribe(new ConsumerGroupedObservableCharacter, String() {Overridepublic void accept(final GroupedObservableCharacter, String characterStringGroupedObservable) throws Exception {//排序后直接订阅输出key和valuecharacterStringGroupedObservable.sorted().subscribe(new ConsumerString() {Overridepublic void accept(String s) throws Exception {System.out.println(onNext key: characterStringGroupedObservable.getKey() value: s);}});}});
}
复制代码输出 onNext key:A value:Android
onNext key:c value:c
onNext key:c value:c#
onNext key:c value:c
onNext key:j value:java
onNext key:j value:javaScript
复制代码4、scan scan操作符会对发射的数据和上一轮发射的数据进行函数处理并返回的数据供下一轮使用持续这个过程来产生剩余的数据流。其应用场景有简单的累加计算判断所有数据的最小值等 public static void main(String[] args) {Observable.just(8, 2, 13, 1, 15).scan(new BiFunctionInteger, Integer, Integer() {Overridepublic Integer apply(Integer integer, Integer integer2) throws Exception {return integer integer2 ? integer : integer2;}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer item) throws Exception {System.out.println(onNext item);}});
}
复制代码输出 onNext8
onNext2
onNext2
onNext1
onNext1
复制代码5、buffer buffer操作符可以将发射出来的数据流在给定的缓存池中进行缓存当缓存池中的数据项溢满时则将缓存池的数据项进行输出重复上述过程直到将发射出来的数据全部发射出去。如果发射出来的数据不够缓存池的大小则按照当前发射出来的数量进行输出。如果对buffer操作符设置了skip参数则buffer每次缓存池溢满时会跳过指定的skip数据项然后再进行缓存和输出。 Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9).buffer(5).subscribe(new ConsumerListInteger() {Overridepublic void accept(ListInteger integers) throws Exception {System.out.println(onNext integers.toString());}
});
复制代码输出 onNext[1, 2, 3, 4, 5]
onNext[6, 7, 8, 9]
复制代码6、window window操作符和buffer操作符在功能上实现的效果是一样的但window操作符最大区别在于同样是缓存一定数量的数据项window操作符最终发射出来的是新的事件流integerObservable而buffer操作符发射出来的是新的数据流也就是说window操作符发射出来新的事件流中的数据项还可以经过Rxjava其他操作符进行处理。 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9).window(2, 1).subscribe(new ConsumerObservableInteger() {Overridepublic void accept(ObservableInteger integerObservable) throws Exception {integerObservable.subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});}});
}
复制代码输出 onNext1
onNext2
onNext2
onNext3
onNext3
onNext4
onNext4
onNext5
onNext5
onNext6
onNext6
onNext7
onNext7
onNext8
onNext8
onNext9
onNext9
复制代码7、小结 map():对数据流的类型进行转换flatMap():对数据流的类型进行包装成另一个数据流groupby():对所有的数据流进行分组scan():对上一轮处理过后的数据流进行函数处理buffer():缓存发射的数据流到一定数量随后发射出数据流集合window():缓存发射的数据流到一定数量随后发射出新的事件流3.3 Filtering Observables (过滤操作符) 1、debounce debounce操作符会去过滤掉发射速率过快的数据项下面的例子onNext事件可以想象成按钮的点击事件如果在2秒种内频繁的点击则其点击事件会被忽略当i为3的除数的时候发射的事件的时间会超过规定忽略事件的时间那么则允许触发点击事件。这就有点像我们频繁点击按钮但始终只会触发一次点击事件这样就不会导致重复去响应点击事件 public static void main(String[] args) {Observable.create(new ObservableOnSubscribeInteger() {Overridepublic void subscribe(ObservableEmitterInteger emitter) throws Exception {for (int i 0; i 100; i) {if (i % 3 0) {Thread.sleep(3000);} else {Thread.sleep(1000);}emitter.onNext(i);}}}).debounce(2, TimeUnit.SECONDS).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext2
onNext5
onNext8
onNext11
onNext14
......
复制代码2、distinct distinct操作符会过滤重复发送的数据项 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 1, 2, 3).distinct().subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
复制代码3、elementAt elementAt操作符只取指定的角标的事件 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 1, 2, 3).elementAt(0).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
复制代码4、filter filter操作符可以过滤指定函数的数据项 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 1, 2, 3).filter(new PredicateInteger() {Overridepublic boolean test(Integer integer) throws Exception {return integer 2;}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext3
onNext4
onNext3
复制代码5、first first操作符只发射第一项数据项 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 1, 2, 3).first(7).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
复制代码6、ignoreElements ignoreElements操作符不发射任何数据只发射事件流的终止通知 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 1, 2, 3).ignoreElements().subscribe(new Action() {Overridepublic void run() throws Exception {System.out.println(onComplete);}});
}
复制代码输出 onComplete
复制代码7、last last操作符只发射最后一项数据 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 1, 2, 3).last(7).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext3
复制代码8、sample sample操作符会在指定的事件内从数据项中采集所需要的数据由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public void sample() {Observable.interval(1, TimeUnit.SECONDS).sample(2, TimeUnit.SECONDS).subscribe(new ConsumerLong() {Overridepublic void accept(Long aLong) throws Exception {System.out.println(onNext aLong);}});
}
复制代码输出 onNext2
onNext4
onNext6
onNext8
复制代码9、skip skip操作符可以忽略事件流发射的前N项数据项只保留之后的数据 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5, 6, 7, 8).skip(3).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer i) throws Exception {System.out.println(onNext i);}});
}
复制代码输出 onNext4
onNext5
onNext6
onNext7
onNext8
复制代码10、skipLast skipLast操作符可以抑制事件流发射的后N项数据 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5, 6, 7, 8).skipLast(3).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer i) throws Exception {System.out.println(onNext i);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onNext5
复制代码11、take take操作符可以在事件流中只发射前面的N项数据 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5, 6, 7, 8).take(3).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer i) throws Exception {System.out.println(onNext i);}});
}
复制代码输出 onNext1
onNext2
onNext3
复制代码12、takeLast takeLast操作符事件流只发射数据流的后N项数据项忽略前面的数据项 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5, 6, 7, 8).takeLast(3).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer i) throws Exception {System.out.println(onNext i);}});
}
复制代码输出 onNext6
onNext7
onNext8
复制代码还有一个操作符叫takeLastBuffer它和takeLast类似唯一的不同是它把所有的数据项收集到一个List再发射而不是依次发射一个 13、小结 debounce():事件流只发射规定范围时间内的数据项distinct():事件流只发射不重复的数据项elementAt():事件流只发射第N个数据项filter():事件流只发射符合规定函数的数据项first():事件流只发射第一个数据项ignoreElements():忽略事件流的发射只发射事件流的终止事件last():事件流只发射最后一项数据项sample():事件流对指定的时间间隔进行数据项的采样skip():事件流忽略前N个数据项skipLast():事件流忽略后N个数据项take():事件流只发射前N个数据项takeLast():事件流只发射后N个数据项3.4 Combining Observables (组合操作符) 1、merge/concat merge操作符可以合并两个事件流如果在merge操作符上增加延时发送的操作那么就会导致其发射的数据项是无序的会跟着发射的时间点进行合并。虽然是将两个事件流合并成一个事件流进行发射但在最终的一个事件流中发射出来的却是两次数据流。由于concat操作符和merge操作符的效果是一样的这里只举一例 merge和concat的区别 merge():合并后发射的数据项是无序的concat():合并后发射的数据项是有序的public static void main(String[] args) {ObservableString just1 Observable.just(A, B, C, D, E);ObservableString just2 Observable.just(1, 2, 3, 4, 5);Observable.merge(just1, just2).subscribe(new ConsumerSerializable() {Overridepublic void accept(Serializable serializable) throws Exception {System.out.println(onNext serializable.toString());}});
}
复制代码输出 onNextA
onNextB
onNextC
onNextD
onNextE
onNext1
onNext2
onNext3
onNext4
onNext5
复制代码2、zip zip操作符是将两个数据流进行指定的函数规则合并 public static void main(String[] args) {ObservableString just1 Observable.just(A, B, C, D, E);ObservableString just2 Observable.just(1, 2, 3, 4, 5);Observable.zip(just1, just2, new BiFunctionString, String, String() {Overridepublic String apply(String s, String s2) throws Exception {return s s2;}}).subscribe(new ConsumerString() {Overridepublic void accept(String s) throws Exception {System.out.println(onNext s);}});
}
复制代码输出 onNextA1
onNextB2
onNextC3
onNextD4
onNextE5
复制代码3、startWith startWith操作符是将另一个数据流合并到原数据流的开头 public static void main(String[] args) {ObservableString just1 Observable.just(A, B, C, D, E);ObservableString just2 Observable.just(1, 2, 3, 4, 5);just1.startWith(just2).subscribe(new ConsumerString() {Overridepublic void accept(String s) throws Exception {System.out.println(onNext s);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onNext5
onNextA
onNextB
onNextC
onNextD
onNextE
复制代码4、join join操作符是有时间期限的合并操作符由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public void join() {ObservableString just1 Observable.just(A, B, C, D, E);ObservableLong just2 Observable.interval(1, TimeUnit.SECONDS);just1.join(just2, new FunctionString, ObservableSourceLong() {Overridepublic ObservableSourceLong apply(String s) throws Exception {return Observable.timer(3, TimeUnit.SECONDS);}}, new FunctionLong, ObservableSourceLong() {Overridepublic ObservableSourceLong apply(Long l) throws Exception {return Observable.timer(8, TimeUnit.SECONDS);}}, new BiFunctionString, Long, String() {Overridepublic String apply(String s, Long l) throws Exception {return s l;}}).subscribe(new ConsumerString() {Overridepublic void accept(String s) throws Exception {System.out.println(onNext s);}});
}
复制代码join操作符有三个函数需要设置 第一个函数规定just2的过期期限第二个函数规定just1的过期期限第三个函数规定just1和just2的合并规则由于just2的期限只有3秒的时间而just2延时1秒发送一次所以just2只发射了2次其输出的结果就只能和just2输出的两次进行合并其输出格式有点类似我们的排列组合 onNextA0
onNextB0
onNextC0
onNextD0
onNextE0
onNextA1
onNextB1
onNextC1
onNextD1
onNextE1
复制代码5、combineLatest conbineLatest操作符会寻找其他事件流最近发射的数据流进行合并由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public static String[] str {A, B, C, D, E};public void combineLatest() {ObservableString just1 Observable.interval(1, TimeUnit.SECONDS).map(new FunctionLong, String() {Overridepublic String apply(Long aLong) throws Exception {return str[(int) (aLong % 5)];}});ObservableLong just2 Observable.interval(1, TimeUnit.SECONDS);Observable.combineLatest(just1, just2, new BiFunctionString, Long, String() {Overridepublic String apply(String s, Long l) throws Exception {return s l;}}).subscribe(new ConsumerString() {Overridepublic void accept(String s) throws Exception {System.out.println(onNext s);}});
}
复制代码输出 onNextA0
onNextB0
onNextB1
onNextC1
onNextC2
onNextD2
onNextD3
onNextE3
onNextE4
onNextA4
onNextA5
复制代码6、小结 merge()/concat():无序/有序的合并两个数据流zip():两个数据流的数据项合并成一个数据流一同发出startWith():将待合并的数据流放在自身前面一同发出join():将数据流进行排列组合发出不过数据流都是有时间期限的combineLatest():合并最近发射出的数据项成数据流一同发出3.5 Error Handling Operators(错误处理操作符) 1、onErrorReturn onErrorReturn操作符表示当错误发生时它会忽略onError的回调且会发射一个新的数据项并回调onCompleted() public static void main(String[] args) {Observable.create(new ObservableOnSubscribeInteger() {Overridepublic void subscribe(NonNull ObservableEmitterInteger e) throws Exception {for (int i 1; i 5; i) {if(i 4){e.onError(new Exception(onError crash));}e.onNext(i);}}}).onErrorReturn(new FunctionThrowable, Integer() {Overridepublic Integer apply(Throwable throwable) throws Exception {return -1;}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}}, new ConsumerThrowable() {Overridepublic void accept(Throwable throwable) throws Exception {System.out.println(onError);}}, new Action() {Overridepublic void run() throws Exception {System.out.println(onComplete);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext-1
onComplete
复制代码2、onErrorResumeNext onErrorResumeNext操作符表示当错误发生时它会忽略onError的回调且会发射一个新的事件流并回调onCompleted() public static void main(String[] args) {Observable.create(new ObservableOnSubscribeInteger() {Overridepublic void subscribe(NonNull ObservableEmitterInteger e) throws Exception {for (int i 1; i 5; i) {if(i 4){e.onError(new Exception(onError crash));}e.onNext(i);}}}).onErrorResumeNext(new FunctionThrowable, ObservableSource? extends Integer() {Overridepublic ObservableSource? extends Integer apply(Throwable throwable) throws Exception {return Observable.just(-1);}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}}, new ConsumerThrowable() {Overridepublic void accept(Throwable throwable) throws Exception {System.out.println(onError);}}, new Action() {Overridepublic void run() throws Exception {System.out.println(onComplete);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext-1
onComplete
复制代码3、onExceptionResumeNext onExceptionResumeNext操作符表示当错误发生时如果onError收到的Throwable不是一个Exception它会回调onError方法且不会回调备用的事件流如果onError收到的Throwable是一个Exception它会回调备用的事件流进行数据的发射 public static void main(String[] args) {Observable.create(new ObservableOnSubscribeInteger() {Overridepublic void subscribe(NonNull ObservableEmitterInteger e) throws Exception {for (int i 1; i 5; i) {if(i 4){e.onError(new Exception(onException crash));//e.onError(new Error(onError crash));}e.onNext(i);}}}).onExceptionResumeNext(new ObservableSourceInteger() {Overridepublic void subscribe(Observer? super Integer observer) {//备用事件流observer.onNext(8);}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}}, new ConsumerThrowable() {Overridepublic void accept(Throwable throwable) throws Exception {System.out.println(onError);}}, new Action() {Overridepublic void run() throws Exception {System.out.println(onComplete);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext8
复制代码4、retry retry操作符表示当错误发生时发射器会重新发射 public static void main(String[] args) {Observable.create(new ObservableOnSubscribeInteger() {Overridepublic void subscribe(NonNull ObservableEmitterInteger e) throws Exception {for (int i 1; i 5; i) {if (i 4) {e.onError(new Exception(onError crash));}e.onNext(i);}}}).retry(1).onErrorReturn(new FunctionThrowable, Integer() {Overridepublic Integer apply(Throwable throwable) throws Exception {return -1;}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}}, new ConsumerThrowable() {Overridepublic void accept(Throwable throwable) throws Exception {System.out.println(onError);}}, new Action() {Overridepublic void run() throws Exception {System.out.println(onComplete);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext1
onNext2
onNext3
onNext-1
onComplete
复制代码retry():表示重试无限次retry(long times):表示重试指定次数retry(Func predicate):可以根据函数参数中的Throwable类型和重试次数决定本次需不需要重试5、retryWhen retryWhen操作符和retry操作符相似区别在于retryWhen将错误Throwable传递给了函数进行处理并产生新的事件流进行处理由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 private static int retryCount 0;
private static int maxRetries 2;public void retryWhen(){Observable.create(new ObservableOnSubscribeInteger() {Overridepublic void subscribe(NonNull ObservableEmitterInteger e) throws Exception {for (int i 1; i 5; i) {if (i 4) {e.onError(new Exception(onError crash));}e.onNext(i);}}}).retryWhen(new FunctionObservableThrowable, ObservableSource?() {Overridepublic ObservableSource? apply(ObservableThrowable throwableObservable) throws Exception {return throwableObservable.flatMap(new FunctionThrowable, ObservableSource?() {Overridepublic ObservableSource? apply(Throwable throwable) throws Exception {if (retryCount maxRetries) {// When this Observable calls onNext, the original Observable will be retried (i.e. re-subscribed).System.out.println(get error, it will try after 1 seconds, retry count retryCount);return Observable.timer(1, TimeUnit.SECONDS);}return Observable.error(throwable);}});}}).onErrorReturn(new FunctionThrowable, Integer() {Overridepublic Integer apply(Throwable throwable) throws Exception {return -1;}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}}, new ConsumerThrowable() {Overridepublic void accept(Throwable throwable) throws Exception {System.out.println(onError);}}, new Action() {Overridepublic void run() throws Exception {System.out.println(onComplete);}});
}
复制代码输出 onNext1
onNext2
onNext3
get error, it will try after 1 seconds, retry count 1
onNext1
onNext2
onNext3
get error, it will try after 1 seconds, retry count 2
onNext1
onNext2
onNext3
onNext-1
onComplete
复制代码6、小结 onErrorReturn():当错误发生时它会忽略onError的回调且会发射一个新的数据项并回调onCompleted()onErrorResumeNext():当错误发生时它会忽略onError的回调且会发射一个新的事件流并回调onCompleted()onExceptionResumeNext():当错误发生时如果onError收到的Throwable不是一个Exception它会回调onError方法且不会回调备用的事件流如果onError收到的Throwable是一个Exception它会回调备用的事件流进行数据的发射retry():当错误发生时发射器会重新发射retryWhen():当错误发生时根据Tharowble类型决定发射器是否重新发射3.6 Observable Utility Operators(辅助性操作符) 1、delay delay操作符可以延时某次事件发送的数据流由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public void deley() {Observable.just(1, 2, 3, 4, 5).delay(2, TimeUnit.SECONDS).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onNext5
复制代码delay和delaySubscription的效果是一样的只不过delay是对数据流的延时而delaySubscription是对事件流的延时 2、do do操作符可以监听整个事件流的生命周期do操作符分为多个类型而且每个类型的作用都不同 doOnNext():接收每次发送的数据项doOnEach():接收每次发送的数据项doOnSubscribe():当事件流被订阅时被调用doOnDispose():当事件流被释放时被调用doOnComplete():当事件流被正常终止时被调用doOnError():当事件流被异常终止时被调用doOnTerminate():当事件流被终止之前被调用无论正常终止还是异常终止都会调用doFinally():当事件流被终止之后被调用无论正常终止还是异常终止都会调用public static void main(String[] args) {Observable.just(1, 2, 3).doOnNext(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(doOnNext);}}).doOnEach(new ConsumerNotificationInteger() {Overridepublic void accept(NotificationInteger integerNotification) throws Exception {System.out.println(doOnEach);}}).doOnSubscribe(new ConsumerDisposable() {Overridepublic void accept(Disposable disposable) throws Exception {System.out.println(doOnSubscribe);}}).doOnDispose(new Action() {Overridepublic void run() throws Exception {System.out.println(doOnDispose);}}).doOnTerminate(new Action() {Overridepublic void run() throws Exception {System.out.println(doOnTerminate);}}).doOnError(new ConsumerThrowable() {Overridepublic void accept(Throwable throwable) throws Exception {System.out.println(doOnError);}}).doOnComplete(new Action() {Overridepublic void run() throws Exception {System.out.println(doOnComplete);}}).doFinally(new Action() {Overridepublic void run() throws Exception {System.out.println(doFinally);}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 doOnSubscribe
doOnNext
doOnEach
onNext1
doOnNext
doOnEach
onNext2
doOnNext
doOnEach
onNext3
doOnEach
doOnTerminate
doOnComplete
doFinally
复制代码3、materialize/dematerialize materialize操作符将发射出的数据项转换成为一个Notification对象而dematerialize操作符则是跟materialize操作符相反这两个操作符有点类似我们Java对象的装箱和拆箱功能 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5).materialize().subscribe(new ConsumerNotificationInteger() {Overridepublic void accept(NotificationInteger integerNotification) throws Exception {System.out.println(onNext integerNotification.getValue());}});Observable.just(1, 2, 3, 4, 5).materialize().dematerialize().subscribe(new ConsumerObject() {Overridepublic void accept(Object object) throws Exception {System.out.println(onNext object.toString());}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onNext5
onNextnull
onNext1
onNext2
onNext3
onNext4
onNext5
复制代码输出的时候materialize会输出多个null是因为null的事件为onCompleted事件而dematerialize把onCompleted事件给去掉了这个原因也可以从图片中看出来 4、serialize serialize操作符可以将异步执行的事件流进行同步操作直到事件流结束 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5).serialize().subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onNext5
复制代码5、timeInterval timeInterval操作符可以将发射的数据项转换为带有时间间隔的数据项由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public void timeInterval(){Observable.interval(2, TimeUnit.SECONDS).timeInterval(TimeUnit.SECONDS).subscribe(new ConsumerTimedLong() {Overridepublic void accept(TimedLong longTimed) throws Exception {System.out.println(onNext longTimed.value() timeInterval longTimed.time());}});
}
复制代码输出 onNext0 timeInterval2
onNext1 timeInterval2
onNext2 timeInterval2
onNext3 timeInterval2
onNext4 timeInterval2
复制代码6、timeout timeout操作符表示当发射的数据项超过了规定的限制时间则发射onError事件这里直接让程序超过规定的限制时间由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public void timeOut(){Observable.interval(2, TimeUnit.SECONDS).timeout(1, TimeUnit.SECONDS).subscribe(new ConsumerLong() {Overridepublic void accept(Long aLong) throws Exception {System.out.println(onNext aLong);}}, new ConsumerThrowable() {Overridepublic void accept(Throwable throwable) throws Exception {System.out.println(onError);}});
}
复制代码输出 onError
复制代码7、timestamp timestamp操作符会给每个发射的数据项带上时间戳由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public void timeStamp() {Observable.interval(2, TimeUnit.SECONDS).timestamp(TimeUnit.MILLISECONDS).subscribe(new ConsumerTimedLong() {Overridepublic void accept(TimedLong longTimed) throws Exception {System.out.println(onNext longTimed.value() timeInterval longTimed.time());}});
}
复制代码输出 onNext0 timeInterval1525755132132
onNext1 timeInterval1525755134168
onNext2 timeInterval1525755136132
onNext3 timeInterval1525755138132
复制代码8、using using操作符可以让你的事件流存在一次性的数据项即用完就将资源释放掉 using操作符接受三个参数 一个用户创建一次性资源的工厂函数一个用于创建一次性事件的工厂函数一个用于释放资源的函数public static class UserBean {String name;int age;public UserBean(String name, int age) {this.name name;this.age age;}
}public static void main(String[] args) {Observable.using(new CallableUserBean() {Overridepublic UserBean call() throws Exception {//从网络中获取某个对象return new UserBean(俊俊俊, 22);}}, new FunctionUserBean, ObservableSource?() {Overridepublic ObservableSource? apply(UserBean userBean) throws Exception {//拿出你想要的资源return Observable.just(userBean.name);}}, new ConsumerUserBean() {Overridepublic void accept(UserBean userBean) throws Exception {//释放对象userBean null;}}).subscribe(new ConsumerObject() {Overridepublic void accept(Object o) throws Exception {System.out.println(onNext o.toString());}});
}
复制代码输出 onNext俊俊俊
复制代码9、to to操作符可以将数据流中的数据项进行集合的转换to操作符分为多个类型而且每个类型的作用都不同 toList():转换成List类型的集合toMap():转换成Map类型的集合toMultimap():转换成一对多(即A类型,ListB类型)的Map类型的集合toSortedList():转换成具有排序的List类型的集合public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5).toList().subscribe(new ConsumerListInteger() {Overridepublic void accept(ListInteger integers) throws Exception {System.out.println(onNext integers.toString());}});
}
复制代码输出 onNext[1, 2, 3, 4, 5]
复制代码10、小结 delay():延迟事件发射的数据项do():监听事件流的生命周期materialize()/dematerialize():对事件流进行装箱/拆箱serialize():同步事件流的发射timeInterval():对事件流增加时间间隔timeout():对事件流增加限定时间timestamp():对事件流增加时间戳using():对事件流增加一次性的资源to():对数据流中的数据项进行集合的转换3.7 Conditional and Boolean Operators(条件和布尔操作符) 1、all all操作符表示对所有数据项进行校验如果所有都通过则返回true否则返回false public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5).all(new PredicateInteger() {Overridepublic boolean test(Integer integer) throws Exception {return integer 0;}}).subscribe(new ConsumerBoolean() {Overridepublic void accept(Boolean aBoolean) throws Exception {System.out.println(onNext aBoolean);}});
}
复制代码输出 onNexttrue
复制代码2、contains contains操作符表示事件流中发射的数据项当中是否包含有指定的数据项 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5).contains(2).subscribe(new ConsumerBoolean() {Overridepublic void accept(Boolean aBoolean) throws Exception {System.out.println(onNext aBoolean);}});
}
复制代码输出 onNexttrue
复制代码3、amb amb操作符在多个事件流中只发射最先发出数据的事件流由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public void amb(){ListObservableInteger list new ArrayList();list.add(Observable.just(1, 2, 3).delay(3, TimeUnit.SECONDS));list.add(Observable.just(4, 5, 6).delay(2, TimeUnit.SECONDS));list.add(Observable.just(7, 8, 9).delay(1, TimeUnit.SECONDS));Observable.amb(list).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext7
onNext8
onNext9
复制代码4、defaultIfEmpty defaultIfEmpty操作符会在事件流没有发射任何数据时发射一个指定的默认值 public static void main(String[] args) {Observable.empty().defaultIfEmpty(-1).subscribe(new ConsumerObject() {Overridepublic void accept(Object o) throws Exception {System.out.println(onNext o.toString());}});
}
复制代码输出 onNext-1
复制代码5、sequenceEqual sequenceEqual操作符可以判断两个数据流是否完全相等 public static void main(String[] args) {ObservableInteger just1 Observable.just(1, 2, 3);ObservableInteger just2 Observable.just(1, 2, 3);Observable.sequenceEqual(just1, just2).subscribe(new ConsumerBoolean() {Overridepublic void accept(Boolean aBoolean) throws Exception {System.out.println(onNext aBoolean);}});
}
复制代码输出 onNexttrue
复制代码6、skipUntil/skipWhile skipUtils操作符是在两个事件流发射的时候第一个事件流会等到第二个事件流开始发射的时候第一个事件流才开始发射出数据项它会忽略之前发射过的数据项由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public void skipUntil(){ObservableLong just1 Observable.interval(1, TimeUnit.SECONDS);ObservableInteger just2 Observable.just(8).delay(3, TimeUnit.SECONDS);just1.skipUntil(just2).subscribe(new ConsumerLong() {Overridepublic void accept(Long aLong) throws Exception {System.out.println(onNext aLong);}});
}
复制代码输出 onNext2
onNext3
onNext4
onNext5
......
复制代码 skipWhile操作符是在一个事件流中从第一项数据项开始判断是否符合某个特定条件如果判断值返回true则不发射该数据项继续从下一个数据项执行同样的判断直到某个数据项的判断值返回false时则终止判断发射剩余的所有数据项。需要注意的是这里只要一次判断为false则后面的所有数据项都不判断 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 5).skipWhile(new PredicateInteger() {Overridepublic boolean test(Integer integer) throws Exception {return integer 3;}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext3
onNext4
onNext5
复制代码7、takeUntil/takeWhile takeUntil操作符跟skipUntil类似skip表示跳过的意思而take表示取值的意思takeUntil操作符是在两个事件流发射的时候第一个事件流会等到第二个事件流开始发射的时候第一个事件流停止发射数据项它会忽略之后的数据项由于这段代码的的延时操作都是非阻塞型的所以在Java上运行会导致JVM的立马停止只能把这段代码放在Android来运行 public void takeUntil(){ObservableLong just1 Observable.interval(1, TimeUnit.SECONDS);ObservableInteger just2 Observable.just(8).delay(3, TimeUnit.SECONDS);just1.takeUntil(just2).subscribe(new ConsumerLong() {Overridepublic void accept(Long aLong) throws Exception {System.out.println(onNext aLong);}});
}
复制代码输出 onNext0
onNext1
复制代码 takeWhile操作符是在一个事件流中从第一项数据项开始判断是否符合某个特定条件如果判断值返回true则发射该数据项继续从下一个数据项执行同样的判断直到某个数据项的判断值返回false时则终止判断且剩余的所有数据项不会发射。需要注意的是这里只要一次判断为false则后面的所有数据项都不判断 public static void main(String[] args) {Observable.just(1, 2, 3, 4, 0).takeWhile(new PredicateInteger() {Overridepublic boolean test(Integer integer) throws Exception {return integer 3;}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
onNext2
复制代码8、小结 all():对所有数据项进行校验contains():所有数据项是否包含指定数据项amb():多个事件流中只发射最先发出的事件流defaultIfEmpty():如果数据流为空则发射默认数据项sequenceEqual():判断两个数据流是否完全相等skipUntil():当两个事件流发射时第一个事件流的数据项会等到第二个事件流开始发射时才进行发射skipWhile():当发射的数据流达到某种条件时才开始发射剩余所有数据项takeUntil():当两个事件流发射时第一个事件流的数据项会等到第二个事件流开始发射时终止发射takeWhile():当发射的数据流达到某种条件时才停止发射剩余所有数据项3.8 Mathematical and Aggregate Operators(数学运算及聚合操作符) 数学运算操作符比较简单对于数学运算操作符会放在小结中介绍下面是对聚合操作符做介绍 1、reduce reduce操作符跟scan操作符是一样的会对发射的数据和上一轮发射的数据进行函数处理并返回的数据供下一轮使用持续这个过程来产生剩余的数据流。reduce与scan的唯一区别在于reduce只输出最后的结果而scan会输出每一次的结果这点从图片中也能看出来 public static void main(String[] args) {Observable.just(8, 2, 13, 1, 15).reduce(new BiFunctionInteger, Integer, Integer() {Overridepublic Integer apply(Integer integer, Integer integer2) throws Exception {return integer integer2 ? integer : integer2;}}).subscribe(new ConsumerInteger() {Overridepublic void accept(Integer item) throws Exception {System.out.println(onNext item);}});
}
复制代码输出 onNext1
复制代码2、collect collect操作符跟reduce操作符类似只不过collect增加了一个可改变数据结构的函数供我们处理 public static void main(String[] args) {Observable.just(8, 2, 13, 1, 15).collect(new CallableString() {Overridepublic String call() throws Exception {return A;}}, new BiConsumerString, Integer() {Overridepublic void accept(String s, Integer integer) throws Exception {System.out.println(onNext s integer);}}).subscribe(new BiConsumerString, Throwable() {Overridepublic void accept(String s, Throwable throwable) throws Exception {System.out.println(onNext2 s);}});
}
复制代码输出 onNextA 8
onNextA 2
onNextA 13
onNextA 1
onNextA 15
onNext2A
复制代码3、小结 数学运算操作符的使用需要在gradle中添加rxjava-math的依赖 implementation io.reactivex:rxjava-math:1.0.0
复制代码average():求所有数据项的平均值max/min():求所有数据项的最大或最小值sum():求所有数据项的总和reduce():对上一轮处理过后的数据流进行函数处理只返回最后的结果collect():对上一轮处理过后的数据流进行函数处理可改变原始的数据结构3.9 Connectable Observable(连接操作符) 1、publish publish操作符是将普通的事件流转化成可连接的事件流ConnectableObservable它与普通的事件流不一样ConnectableObservable在没有调用connect()进行连接的情况下事件流是不会发射数据的 public static void main(String[] args) {ConnectableObservableInteger connectableObservable Observable.just(1, 2, 3, 4, 5).publish();connectableObservable.subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 无
复制代码2、connect connect操作符是将可连接的事件流进行连接并开始发射数据。这个方法需要注意的是connect操作符必须在所有事件流被订阅后才开始发射数据。如果放在subscribe之前的话则订阅者是无法收到数据的。如果后面还有订阅者将订阅此次事件流则会丢失已经调用了connect后发射出去的数据项 public static void main(String[] args) {ConnectableObservableInteger connectableObservable Observable.just(1, 2, 3, 4, 5).publish();connectableObservable.subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});connectableObservable.connect();
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onNext5
复制代码3、refCount refCount操作符可以将可连接的事件流转换成普通的事件流 public static void main(String[] args) {ConnectableObservableInteger connectableObservable Observable.just(1, 2, 3, 4, 5).publish();connectableObservable.refCount().subscribe(new ConsumerInteger() {Overridepublic void accept(Integer integer) throws Exception {System.out.println(onNext integer);}});
}
复制代码输出 onNext1
onNext2
onNext3
onNext4
onNext5
复制代码4、replay replay操作符将弥补connect操作符的缺陷由于connect会让后面进行订阅的订阅者丢失之前发射出去的数据项所以使用replay操作符可以将发射出去的数据项进行缓存这样使得后面的订阅者都可以获得完整的数据项。这里需要注意的是replay操作符不能和publish操作符同时使用否则将不会发射数据。例子中读者可以将replay操作符换成publish操作符这时候的输出就会丢失前2秒发射的数据项 public void replay(){ConnectableObservableLong connectableObservable Observable.interval(1, TimeUnit.SECONDS).replay();connectableObservable.connect();connectableObservable.delaySubscription(3, TimeUnit.SECONDS).subscribe(new ConsumerLong() {Overridepublic void accept(Long aLong) throws Exception {System.out.println(onNext aLong);}});
}
复制代码输出 onNext0
onNext1
onNext2
onNext3
onNext4
onNext5
......
复制代码5、小结 publish():将普通的事件流转换成可连接的事件流connect():将可连接的事件流进行连接并发射数据refCount():将可连接的事件流转换成普通的事件流replay():缓存可连接的事件流中的所有数据项转载于:https://juejin.im/post/5cd8dc55f265da039f0f3030
