介紹
在RxJava的事件流轉(zhuǎn)過(guò)程中窖式，可以改變事件中的事件以及數(shù)據(jù)怀吻，使用的就是RxJava提供的操作符。
從RxJava2-Android-Samples的ReadMe.md中可以看到有如下操作符
| 操作符 | 操作符 | 操作符 | 操作符 |
|---|---|---|---|
| Map | Zip | Reduce | Filter |
| FlatMap | Take | Skip | Buffer |
| Concat | Replay | Merge |
而在Operators的文檔中圖文并茂展示出各個(gè)操作符的作用接下來(lái)會(huì)介紹這些操作符的作用與使用方法
Map
Map的作用：
將發(fā)射的事件進(jìn)行Map函數(shù)定義的數(shù)據(jù)轉(zhuǎn)換，再將轉(zhuǎn)換后的事件發(fā)射給Observer洼裤。
如下示例：
- 通過(guò)
Emitter發(fā)射1，2肛鹏，3三個(gè)事件 - 中間會(huì)通過(guò)Map進(jìn)行轉(zhuǎn)換逸邦，轉(zhuǎn)換后的事件為
10恩沛，20，30 - 再將轉(zhuǎn)換后的事件發(fā)送給
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操作符的作用：
將多個(gè)發(fā)射器的事件按順序合并成單個(gè)事件流發(fā)射缕减，如果多個(gè)發(fā)射器之間的事件數(shù)不相等雷客，則合并后的事件數(shù)以少的為準(zhǔn)。
如下示例：
- 通過(guò)
Observable.interval創(chuàng)建0桥狡，1搅裙，2這三個(gè)事件的發(fā)射器 - 通過(guò)
Observable.interval創(chuàng)建a，b裹芝，c部逮，d，e對(duì)應(yīng)ASCII碼的發(fā)射器嫂易，并且通過(guò)filter過(guò)濾其他的ASCII碼值兄朋，因?yàn)?code>a的ASCII是97，通過(guò)take獲取前面6個(gè)元素 - 調(diào)用
Observable.zip創(chuàng)建一個(gè)合并了nums和chars的發(fā)射器怜械，并且通過(guò)BiFunction將兩個(gè)發(fā)射器發(fā)射的數(shù)據(jù)進(jìn)行合并 - 通過(guò)合并后的發(fā)射器颅和，調(diào)用
subscribe來(lái)接收事件
// 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")
}
打印出來(lái)的日志如下：
Subscribe...0....a
Subscribe...1....b
Subscribe...2....c
注釋：
-
Observable.interval是根據(jù)傳入的參數(shù)生成順序數(shù)組的方法。如上例所示缕允，每隔250毫秒就會(huì)發(fā)送一個(gè)從0開(kāi)始的事件峡扩，發(fā)送0，1障本，2教届，3...一直累加的事件 -
zip只會(huì)以最少的發(fā)射器數(shù)量為標(biāo)準(zhǔn)合并事件。如上例所示驾霜，nums發(fā)射器發(fā)送0,1,2這3個(gè)事件案训，而chars發(fā)送a，b寄悯，c萤衰，d，e猜旬，f這5個(gè)事件脆栋，但是合并過(guò)后，只會(huì)產(chǎn)生3個(gè)事件
Reduce
Reduce本意是減少的意思洒擦，而它的作用也就是將發(fā)射器的事件都合并到一個(gè)事件中椿争，以減少事件的接收。
如下示例：
- 每250毫秒發(fā)送一個(gè)事件
- 在
reduce接收的函數(shù)中熟嫩，每次將發(fā)送的事件結(jié)果與之前的事件結(jié)果相加秦踪，并且返回 - 在
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名為過(guò)濾，與其名相似，Filter的作用就是按條件過(guò)濾事件椅邓，只有返回true的事件才能繼續(xù)往Observer傳遞
如上圖所示柠逞，如果發(fā)送的數(shù)大于10，則允許發(fā)射景馁，否則會(huì)被過(guò)濾
FlatMap
FlatMap名為扁平映射板壮，它的作用就是將發(fā)射端的事件按照FlatMap中定義的策略進(jìn)行拆分，拆分成多個(gè)事件后合住，繼續(xù)向下傳遞事件
如下示例：
在最開(kāi)始的發(fā)射端绰精，發(fā)射0，1透葛，2笨使，而經(jīng)過(guò)了flatMap之后，會(huì)將每個(gè)接收的事件進(jìn)行處理僚害，處理后進(jìn)行二次發(fā)射硫椰。也就是將0這個(gè)事件轉(zhuǎn)換成了Receive:0以及Emitter:100這兩個(gè)事件往下發(fā)送，所以贡珊，在Observer端接收到的事件就是這兩個(gè)事件了最爬。
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的作用比較簡(jiǎn)單，指的就是獲取前面N個(gè)事件门岔，后面的事件則會(huì)直接拋棄
而TakeLast的作用則是獲取后面的N個(gè)事件，而前面的事件會(huì)直接拋棄
Skip與SkipLast
Skip的作用就是跳過(guò)N個(gè)事件
SkipLast的作用則就是跳過(guò)后面的N個(gè)事件烤送，例如發(fā)送了4個(gè)事件寒随，1，2帮坚，3妻往，4，而SkipLast(2)试和，則會(huì)跳過(guò)后面的3讯泣，4，只發(fā)送1阅悍，2好渠。而如果只發(fā)送2個(gè)事件的話，則SkipLast(2)調(diào)用后节视，則不會(huì)發(fā)送任何事件拳锚。
Buffer
Buffer即為緩沖，會(huì)將發(fā)射的事件積攢到count數(shù)量后寻行，統(tǒng)一發(fā)送給Observer霍掺，而Obserer端接收到到則是一個(gè)MutableList。如果發(fā)射到最后，即使沒(méi)有滿足條件杆烁，也會(huì)將剩下到打包發(fā)送
如下示例：
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可以做到多個(gè)發(fā)射器并行發(fā)射事件牙丽，而在接收端會(huì)在接收完前一個(gè)發(fā)射器的事件后，再接收下一個(gè)發(fā)射器的事件兔魂，接收順序不會(huì)交錯(cuò)烤芦。
如下例子：
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的作用也就是在事件已經(jīng)被前一個(gè)Observer接收后，當(dāng)下一個(gè)Observer重新Subscribe時(shí)入热，仍然可以將之前的事件傳遞給新的Obserer拍棕，有點(diǎn)類似于Sticky
如下示例：
創(chuàng)建一個(gè)ReplaySubject，然后在第一個(gè)Observer中打印First Subscribe勺良，發(fā)射完1绰播，2，3尚困，4之后蠢箩，再重新讓一個(gè)新的Observer訂閱該發(fā)射器，在第二個(gè)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")
}
輸出結(jié)果如下：
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的作用可以讓多個(gè)發(fā)射器同時(shí)發(fā)射谬泌，并且在Observer端接收時(shí)，按發(fā)射事件的時(shí)間順序來(lái)接收
如下示例：
val emitter1 = Observable.fromArray(0, 1, 2)
val emitter2 = Observable.fromArray('a', 'b', 'c')
Observable.merge(emitter1, emitter2).subscribe {
Log.e(TAG, "Subscribe...$it")
}
輸出的結(jié)果：
E/SelectImageActivity: Subscribe...0
E/SelectImageActivity: Subscribe...1
E/SelectImageActivity: Subscribe...2
E/SelectImageActivity: Subscribe...a
E/SelectImageActivity: Subscribe...b
E/SelectImageActivity: Subscribe...c
