RxJava2--操作符Operator

介绍

在RxJava的事件流转过程中，可以改变事件中的事件以及数据，使用的就是RxJava提供的操作符。

从RxJava2-Android-Samples的ReadMe.md中可以看到有如下操作符

操作符	操作符	操作符	操作符
Map	Zip	Reduce	Filter
FlatMap	Take	Skip	Buffer
Concat	Replay	Merge

而在Operators的文档中图文并茂展示出各个操作符的作用接下来会介绍这些操作符的作用与使用方法

Map

Map的作用：
将发射的事件进行Map函数定义的数据转换，再将转换后的事件发射给Observer。

Map

如下示例：

1. 通过Emitter发射1，2，3三个事件
2. 中间会通过Map进行转换，转换后的事件为10，20，30
3. 再将转换后的事件发送给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操作符的作用：
将多个发射器的事件按顺序合并成单个事件流发射，如果多个发射器之间的事件数不相等，则合并后的事件数以少的为准。

Zip

如下示例：

1. 通过Observable.interval创建0，1，2这三个事件的发射器
2. 通过Observable.interval创建a，b，c，d，e对应ASCII码的发射器，并且通过filter过滤其他的ASCII码值，因为a的ASCII是97，通过take获取前面6个元素
3. 调用Observable.zip创建一个合并了nums和chars的发射器，并且通过BiFunction将两个发射器发射的数据进行合并
4. 通过合并后的发射器，调用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

注释：

1. Observable.interval是根据传入的参数生成顺序数组的方法。如上例所示，每隔250毫秒就会发送一个从0开始的事件，发送0，1，2，3...一直累加的事件
2. zip只会以最少的发射器数量为标准合并事件。如上例所示，nums发射器发送0,1,2这3个事件，而chars发送a，b，c，d，e，f这5个事件，但是合并过后，只会产生3个事件

Reduce

Reduce本意是减少的意思，而它的作用也就是将发射器的事件都合并到一个事件中，以减少事件的接收。

Reduce

如下示例：

1. 每250毫秒发送一个事件
2. 在reduce接收的函数中，每次将发送的事件结果与之前的事件结果相加，并且返回
3. 在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传递

Filter

如上图所示，如果发送的数大于10，则允许发射，否则会被过滤

FlatMap

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个事件，后面的事件则会直接抛弃

Take

而TakeLast的作用则是获取后面的N个事件，而前面的事件会直接抛弃

TakeLast

Skip与SkipLast

Skip的作用就是跳过N个事件

Skip

SkipLast的作用则就是跳过后面的N个事件，例如发送了4个事件，1，2，3，4，而SkipLast(2)，则会跳过后面的3，4，只发送1，2。而如果只发送2个事件的话，则SkipLast(2)调用后，则不会发送任何事件。

SkipLast

Buffer

Buffer即为缓冲，会将发射的事件积攒到count数量后，统一发送给Observer，而Obserer端接收到到则是一个MutableList。如果发射到最后，即使没有满足条件，也会将剩下到打包发送

Buffer

如下示例：

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可以做到多个发射器并行发射事件，而在接收端会在接收完前一个发射器的事件后，再接收下一个发射器的事件，接收顺序不会交错。

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

image.png

如下示例：
创建一个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端接收时，按发射事件的时间顺序来接收

Merge

如下示例：

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