當 Kotlin 遇見 RxJava 多數據源
溫馨提醒
閱讀本文最好有Kotlin基礎，若沒有基礎，可參考之前文章Kotlin初探， 使用Kotlin優雅的開發Android應用 ，以及RxJava基礎（本文基于RxJava2），當然我也會盡可能詳細解釋讓你順利閱讀本文。
寫在前面
最近幾天回過頭，看了之前的總結RxJava操作符系列，感覺對Rxjava多數據源的處理不是很理解，所以在總結學習一波。大家都知道，最近Kotlin語言一直占據熱搜榜，褒貶不一，但我想說，不管有什么想法都要拋在腦后，畢竟Google爸爸出手，你不情愿也要跟隨它的步伐。鑒于此，本篇對RxJava多數據源的總結是基于Kotlin語言，也讓大家明白，使用Kotlin開發應用并不是不能使用Java庫，現在有一部分人擔心，Kotlin第三方庫那么少，如果使用Kotlin開發那不是給自己找罪受，其實你完全錯了，當你說這話的時候，我敢斷定你都還沒有接觸Kotlin，因為Koltin有一個最重要的優勢就是和Java絕對兼容。
多數據源處理操作符
在RxJava中多數據源處理的操作符很多，但是最經典的就要數merge,contact，zip了。如果對這三個操作符不是很熟悉的話，可以去查看它的使用，當然如果你懶得去看，我也會簡單提一下。merge操作符可以處理多個Observable發送的數據，它是一個異步操作，不保證數據發送的順序，即有可能出現數據交叉，當一個Observable發送了onError后，未執行的Observable不在繼續執行，直接執行merge的onError方法。
contact操作符執行時一個同步操作，嚴格按照contact中傳入Observable先后執行，即前面的先執行后面的后執行，并且最終發送的數據也是有序的，即第一個Observable的數據發送完畢再發送第二個，依次類推。
zip操作符和contact和merge有了本質的區別，它會將每個Observable個數據項分布對應返回一個Observable再發送，最終發送的數據量與最小數據長度相同。
使用場景分析
假如現在我們有三種商品，有一個查詢商品信息的接口，根據接口可以查詢該商品的價格以及出售地點。商品實體類
data class Goods(var id:Int,var price: Int, var address: String)
在Kotlin語言中，實體類創建用data class 關鍵詞，我們不需要和Java一樣創建get/set方法，只需一行代碼搞定。
創建模擬網絡請求
object NetRequest {
//模擬網絡請求
fun getGoodsObservable(id: Int): Observable<Goods> {
fun getGoodsObservable(id: Int): Observable<Goods> {
return Observable.create {
source ->
Thread.sleep(Random().nextInt(1000).toLong())
var data = Goods(id, Random().nextInt(20), "地址${id}")
source.onNext(data)
source.onComplete()
Log.e("getGoodsObservable:", "${id}")
}
}
}
在上面我們創建了一個單例類，在Kotlin中使用object修飾類時即給我們自動創建了一個單例對象。在每一句代碼結尾我們不需要再和Java一樣寫一個分號“；”來結束，什么也不用寫。
Observable.create使用的是lambda表達式，在Kotlin語言中是支持lambda表達式的。source 就是ObservableEmitter ，所以我們可以調用onNext發送數據。為了更準確的模擬網絡請求，使用Thread.sleep隨機的延遲，模擬網絡請求的時間。
fun getGoodsObservable(id: Int): Observable<Goods> {
return Observable.create {
source ->
Thread.sleep(Random().nextInt(1000).toLong())
var data = Goods(id, Random().nextInt(20), "地址${id}")
source.onNext(data)
source.onComplete()
Log.e("getGoodsObservable:", "${id}")
}
當然由于subscribe只有一個參數，所以我們也可以這樣寫。也就是省略了source ->，此時it就表示該參數數據。
return Observable.create {
Thread.sleep(Random().nextInt(1000).toLong())
var data = Goods(id, Random().nextInt(20), "地址${id}")
it.onNext(data)
it.onComplete()
Log.e("getGoodsObservable:", "${id}")
}
在java中實現如下
return Observable.create(new ObservableOnSubscribe<Goods>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Goods> e) throws Exception {
//處理邏輯
}
});</code></pre>
merge
準備好了請求操作，開始使用merge看看執行的效果。
fun executeMerge() {
Observable.merge(getGoodsObservable(1).subscribeOn(Schedulers.newThread()),
getGoodsObservable(2).subscribeOn(Schedulers.newThread()),
getGoodsObservable(3).subscribeOn(Schedulers.newThread()))
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.toList()
.subscribe({
Log.e(TAG, it.toString())
}, {
Log.e(TAG, it.toString())
})
}
merge中有三個網絡請求操作，并通過subscribeOn(Schedulers.newThread())將網絡請求切換到線程中執行，數據都請求成功后，再通過observeOn(AndroidSchedulers.mainThread())切換到主線程請求數據。為了三請求都成功后，我們在更新UI，所以通過toList()將請求的數據轉換成List一塊發送。在上面的subscribe依然使用的lambda表達式，subscribe（{}，{}）中第一個括號是onSuccess回調，里面的it是接收到的List< Goods >數據，第二個括號是onError回調，it表示異常Throwable對象。
subscribe部分Java代碼
.subscribe(new Consumer<List<Goods>>() {
@Override
public void accept(@NonNull List<Goods> goodses) throws Exception {
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
}
});</code></pre>
當然如果你想使用RxJava2中onSubscribe(@NonNull Disposable d) ，你可以這樣使用subscribe
.subscribe(object : SingleObserver<List<Goods>> {
override fun onSubscribe(d: Disposable?) {
}
override fun onError(e: Throwable?) {
}
override fun onSuccess(t: List<Goods>?) {
}
})
為了觀察，我們將請求成功的數據顯示在界面上，我們創建一個Button，TextView。
class MainActivity : AppCompatActivity(), View.OnClickListener {
val TAG = "MainActivity"
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
setContentView(R.layout.activity_main)
setSupportActionBar(toolbar)
//加入這句import kotlinx.android.synthetic.main.activity_main.*
//不用再findViewById,可直接使用
merge.setOnClickListener(this)
}
override fun onClick(v: View) {
when (v.id) {
R.id.merge -> {
executeMerge()
}
}
//when 關鍵字和Java中的Switch關鍵詞是類似的，
//只不過它比Java中的Switch強大的多，可以接收任何參數，
//然后判斷使用,也可以如下使用
when (v) {
merge -> {
}
}
}
}
contact
我們點擊執行幾次發現，返回的List的數據并不是按照merge參數的先后順序執行的，它是并發的，最終的順序，是由網絡請求的快慢決定的，請求返回數據越快也就表示該數據最早發送，即在List中最靠前。那么此時出現一個問題，如果我想返回數據的List順序嚴格按照位置的先后順序呢？那此時使用merge的話，是不太現實了。當然前面我們提到contact可以使用。那么直接將merge更改為contact執行以下試試，
fun executeContact() {
Observable.concat(getGoodsObservable(1).subscribeOn(Schedulers.newThread()),
getGoodsObservable(2).subscribeOn(Schedulers.newThread()),
getGoodsObservable(3).subscribeOn(Schedulers.newThread()))
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.toList()
.subscribe({
Log.e(TAG, it.toString())
}, {
Log.e(TAG, it.toString())
})
}
的確，發現無論執行多少次List的數據都能按照contact中Observable順序發送，我們想要的效果可以實現了，不過你會發現，效率太差了，這是同步執行啊，只有第一個請求成功，才會去請求第二個，然后第三個，假如一次請求需要一秒，那三次請求至少三秒啊，不能忍。
zip
鑒于上面兩種方式的利弊，如果我們既想如merge一樣并發執行，又想和contact一樣保證順序，是不是有點強迫癥的意思，當然強大的zip就能實現我們想要的效果。如下實現。.
fun executeZip() {
Observable.zip(getGoodsObservable(1),
getGoodsObservable(2),
getGoodsObservable(3),
Function3<Goods, Goods, Goods, List<Goods>>
{ goods0, goods1, goods2 ->
val list = ArrayList<Goods>()
list.add(goods0)
list.add(goods1)
list.add(goods2)
list
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe({
Log.e(TAG, it.toString())
}, {
Log.e(TAG, it.toString())
})
}
既然實現了，那我們運行幾次，發現完美的實現了我們想要的效果，即并發的執行了，也保證了我們請求數據的順序性。
在回調中運用RxJava
在上面我們的單個網絡請求是一個同步的請求，如果我們的網絡請求封裝了，在線程中請求，請求成功后在主線程中回調，那我們又該如何創建呢使用呢？
先來模擬一個子線程請求網絡，請求成功回調數據給主線程。
fun getGoods(ctx:Context,id: Int,callbacks:(goods:Goods)->Unit): Unit {
ctx.doAsync {
Thread.sleep(Random().nextInt(1000).toLong())
var data = Goods(id, Random().nextInt(20), "地址${id}")
ctx.runOnUiThread {
callbacks(data)
}
}
}
getGoods傳了三個參數，第一個Context對象，第二個是商品ID,第三個參數是一個函數，(goods:Goods)->Unit表示第三個參數的類型是一個參數為Goods類型并且返回Unit的函數。使用doAsync 模擬異步請求，請求成功后runOnUiThread 切換到UI線程。然后callbacks（data）將數據回調。這種使用方式比Java中回調優美好用太多了。
接下來就開始在回調成功后創建Observable
fun getGoodsCallBack(id: Int): Observable<Goods> {
var subscrbe: ObservableEmitter<Goods>? = null
var o = Observable.create<Goods> {
subscrbe = it
}
//Kotlin特性
getGoods(this@MainActivity, id) {
subscrbe?.onNext(it)
}
return o
}
fun executeZipCallBack() {
Observable.zip(getGoodsCallBack(1).subscribeOn(Schedulers.newThread()),
getGoodsCallBack(2).subscribeOn(Schedulers.newThread()),
getGoodsCallBack(3).subscribeOn(Schedulers.newThread()),
Function3<Goods, Goods, Goods, List<Goods>>
{ goods0, goods1, goods2 ->
val list = ArrayList<Goods>()
list.add(goods0)
list.add(goods1)
list.add(goods2)
list
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe({
Log.e(TAG, it.toString())
}, {
Log.e(TAG, it.toString())
})
}
ok,到這里回調情況下創建使用RxJava也介紹完畢，到此本篇文章就結束了，有問題歡迎指出，內容雜亂，多多擔待，Hava a wonderful day.
來自：https://juejin.im/entry/5925b8bcda2f60005d80e179