介绍
在RxJava的事件流转过程中，可以改变事件中的事件以及数据，使用的就是RxJava提供的操作符。
从RxJava2-Android-Samples的ReadMe.md中可以看到有如下操作符
| 操作符 | 操作符 | 操作符 | 操作符 |
|---|---|---|---|
| Map | Zip | Reduce | Filter |
| FlatMap | Take | Skip | Buffer |
| Concat | Replay | Merge |
而在Operators的文档中图文并茂展示出各个操作符的作用接下来会介绍这些操作符的作用与使用方法
Map
Map的作用：
将发射的事件进行Map函数定义的数据转换，再将转换后的事件发射给Observer。
如下示例：
- 通过
Emitter发射1，2，3三个事件 - 中间会通过Map进行转换，转换后的事件为
10，20，30 - 再将转换后的事件发送给
Observer
Observable.create<Int> { emitter ->
emitter.onNext(1)
emitter.onNext(2)
emitter.onNext(3)
}.map { value ->
value * 10
}.subscribe { data ->
Log.e(TAG, "Consumer onNext...$data")
}
Zip
Zip操作符的作用：
将多个发射器的事件按顺序合并成单个事件流发射，如果多个发射器之间的事件数不相等，则合并后的事件数以少的为准。
如下示例：
- 通过
Observable.interval创建0，1，2这三个事件的发射器 - 通过
Observable.interval创建a，b，c，d，e对应ASCII码的发射器，并且通过filter过滤其他的ASCII码值，因为a的ASCII是97，通过take获取前面6个元素 - 调用
Observable.zip创建一个合并了nums和chars的发射器，并且通过BiFunction将两个发射器发射的数据进行合并 - 通过合并后的发射器，调用
subscribe来接收事件
// Genertate values 0,1,2
val nums = Observable.interval(250, TimeUnit.MILLISECONDS).take(3)
//Generate values a,b,c,d,e,f.....a的ASCII是97
val chars = Observable.interval(150, TimeUnit.MILLISECONDS).filter { i -> i >= 97 }.take(6)
val newObservable = Observable.zip(nums, chars, BiFunction<Long, Long, String> { x, y -> "$x....${y.toChar()}" })
newObservable.subscribe { value ->
Log.e(TAG, "Subscribe...$value")
}
打印出来的日志如下：
Subscribe...0....a
Subscribe...1....b
Subscribe...2....c
注释：
-
Observable.interval是根据传入的参数生成顺序数组的方法。如上例所示，每隔250毫秒就会发送一个从0开始的事件，发送0，1，2，3...一直累加的事件 -
zip只会以最少的发射器数量为标准合并事件。如上例所示，nums发射器发送0,1,2这3个事件，而chars发送a，b，c，d，e，f这5个事件，但是合并过后，只会产生3个事件
Reduce
Reduce本意是减少的意思，而它的作用也就是将发射器的事件都合并到一个事件中，以减少事件的接收。
如下示例：
- 每250毫秒发送一个事件
- 在
reduce接收的函数中，每次将发送的事件结果与之前的事件结果相加，并且返回 - 在
subscribe中订阅最终的事件
Observable.interval(250, TimeUnit.MILLISECONDS).take(6).reduce { t1: Long?, t2: Long? ->
Log.e(TAG, "t1:$t1....t2:$t2")
t1?.plus(t2 ?: 0)
}.subscribe { value ->
Log.e(TAG, "subscribe $value")
}
打印日志如下：
t1:0....t2:1
t1:1....t2:2
t1:3....t2:3
t1:6....t2:4
t1:10....t2:5
subscribe 15
Filter
Filter名为过滤，与其名相似，Filter的作用就是按条件过滤事件，只有返回true的事件才能继续往Observer传递
如上图所示，如果发送的数大于10，则允许发射，否则会被过滤
FlatMap
FlatMap名为扁平映射，它的作用就是将发射端的事件按照FlatMap中定义的策略进行拆分，拆分成多个事件后，继续向下传递事件
如下示例：
在最开始的发射端，发射0，1，2，而经过了flatMap之后，会将每个接收的事件进行处理，处理后进行二次发射。也就是将0这个事件转换成了Receive:0以及Emitter:100这两个事件往下发送，所以，在Observer端接收到的事件就是这两个事件了。
Observable.fromArray(0, 1, 2).flatMap { value ->
Observable.create<String> { emitter ->
emitter.onNext("Receive:$value")
emitter.onNext("Emitter:${value + 100}")
emitter.onComplete()
}
}.subscribe {
Log.e(TAG, "Subscribe:$")
}
打印的日志如下：
E/SelectImageActivity: Subscribe:Receive:0
E/SelectImageActivity: Subscribe:Emitter:100
E/SelectImageActivity: Subscribe:Receive:1
E/SelectImageActivity: Subscribe:Emitter:101
E/SelectImageActivity: Subscribe:Receive:2
E/SelectImageActivity: Subscribe:Emitter:102
Take与TakeLast
Take的作用比较简单，指的就是获取前面N个事件，后面的事件则会直接抛弃
而TakeLast的作用则是获取后面的N个事件，而前面的事件会直接抛弃
Skip与SkipLast
Skip的作用就是跳过N个事件
SkipLast的作用则就是跳过后面的N个事件，例如发送了4个事件，1，2，3，4，而SkipLast(2)，则会跳过后面的3，4，只发送1，2。而如果只发送2个事件的话，则SkipLast(2)调用后，则不会发送任何事件。
Buffer
Buffer即为缓冲，会将发射的事件积攒到count数量后，统一发送给Observer，而Obserer端接收到到则是一个MutableList。如果发射到最后，即使没有满足条件，也会将剩下到打包发送
如下示例：
Observable.fromArray(0, 1, 2, 3, 4).buffer(3).subscribe { list ->
Log.e(TAG, "Receive List:$list")
list.forEach { item ->
Log.e(TAG, "Receive Item:$item")
}
}
输出如下：
E/SelectImageActivity: Receive List:[0, 1, 2]
E/SelectImageActivity: Receive Item:0
E/SelectImageActivity: Receive Item:1
E/SelectImageActivity: Receive Item:2
E/SelectImageActivity: Receive List:[3, 4]
E/SelectImageActivity: Receive Item:3
E/SelectImageActivity: Receive Item:4
Concat
Concat可以做到多个发射器并行发射事件，而在接收端会在接收完前一个发射器的事件后，再接收下一个发射器的事件，接收顺序不会交错。
如下例子：
val emitter1 = Observable.fromArray(0, 1, 2)
val emitter2 = Observable.fromArray('a', 'b', 'c')
Observable.concat(emitter1, emitter2).subscribe {
Log.e(TAG, "Subscribe...$it")
}
输出的日志如下：
E/SelectImageActivity: Subscribe...0
E/SelectImageActivity: Subscribe...1
E/SelectImageActivity: Subscribe...2
E/SelectImageActivity: Subscribe...a
E/SelectImageActivity: Subscribe...b
E/SelectImageActivity: Subscribe...c
Replay
Replay的作用也就是在事件已经被前一个Observer接收后，当下一个Observer重新Subscribe时，仍然可以将之前的事件传递给新的Obserer，有点类似于Sticky
如下示例：
创建一个ReplaySubject，然后在第一个Observer中打印First Subscribe，发射完1，2，3，4之后，再重新让一个新的Observer订阅该发射器，在第二个Observer中打印Second Subscribe。
val replaySubject = ReplaySubject.create<Int>()
replaySubject.subscribe {
Log.e(TAG, "First Subscribe:$it")
}
replaySubject.onNext(1)
replaySubject.onNext(2)
replaySubject.onNext(3)
replaySubject.onNext(4)
replaySubject.subscribe {
Log.e(TAG, "Second Subscribe:$it")
}
输出结果如下：
First Subscribe:1
First Subscribe:2
First Subscribe:3
First Subscribe:4
Second Subscribe:1
Second Subscribe:2
Second Subscribe:3
Second Subscribe:4
Merge
Merge的作用可以让多个发射器同时发射，并且在Observer端接收时，按发射事件的时间顺序来接收
如下示例：
val emitter1 = Observable.fromArray(0, 1, 2)
val emitter2 = Observable.fromArray('a', 'b', 'c')
Observable.merge(emitter1, emitter2).subscribe {
Log.e(TAG, "Subscribe...$it")
}
输出的结果：
E/SelectImageActivity: Subscribe...0
E/SelectImageActivity: Subscribe...1
E/SelectImageActivity: Subscribe...2
E/SelectImageActivity: Subscribe...a
E/SelectImageActivity: Subscribe...b
E/SelectImageActivity: Subscribe...c
