上個月做的事情比較多：改改iOS bug拷恨，學(xué)python超陆，把項目重構(gòu)成MVP书蚪，深入使用Rxjava喇澡。

這次來說說Rxjava，通過還原一個真實的開發(fā)過程殊校，來感受下rxjava的便利之處晴玖。

巨坑從來都是由小坑慢慢塌陷的

先來看下一段最普通的代碼

rx01.png

在沒有特殊需求的情況下，代碼就這么簡單箩艺。你可以理解為窜醉，獲取一個目錄下的所有文件，將它們一個個傳到服務(wù)器上去艺谆。

看起來好像是沒什么問題榨惰，一個for循環(huán)搞定。一個task失敗了不影響另一個task静汤。每個task run在一個單獨的子線程琅催。

之前rxjava使用場景只局限于和Retrofit一起用。沒過多的使用操作符虫给。因此在uploadFile(path)方法中就是最簡單的Retrofit＋Rxjava上傳文件藤抡。rxjava就切換了下線程。

對于寫慣java的人抹估，這么寫是沒什么問題的缠黍。但如果深入使用過rxjava之后，這么寫就非常別扭了药蜻〈墒剑看到for loop了替饿，你不想將它改成Observable.from()嘛？

把能看見的都改成stream吧

getFileList()方法是獲取sd卡中data包下所有以loc為后綴的files贸典。

workflow分三步：

1. locate to data dir
2. list files under data dir
3. filter files with .loc suffix

換成rxjava非常容易

1. 先發(fā)射一個data目錄路徑
2. 需求是多次上傳文件视卢，得用flatMap將data映射成一個Observable<File[]>
   2.1 當(dāng)然你可以選擇直接listFile(filter)，但這樣回調(diào)又套回調(diào)廊驼，不是很好看据过。
   2.2 用filter操作符將發(fā)射來的File[]過濾

比如像2.1這樣寫

rx02.png

或者像2.2這樣寫

rx03.png

注意，在flatMap中又用from()操作符將File[]變換成一個OnSubscribeFromArray類型的Observable在內(nèi)部通過for循環(huán)一個個發(fā)射妒挎。（感謝一位朋友指正绳锅，之前理解錯誤，以為是發(fā)射多個Observable饥漫。不看源碼真是稀里糊涂的）

假如你的API接口可以接收多個文件榨呆，其實也不用這樣寫。直接在flatMap中拼接RequestBody庸队，調(diào)用API請求就可以了。比如像下圖這樣寫：

rx04.png

無奈需求是上傳loc文件同時還會再帶上一個sensor文件闯割，所以就不能像上述這樣寫彻消。

產(chǎn)品說：需求變了～

接下來的workflow就很有趣了。現(xiàn)在有了多個Observable<File>宙拉，一個個上傳就是了宾尚。

如果不考慮隊列，不考慮無網(wǎng)或上傳失敗情況谢澈。完全再來一個flatMap將Observable<File>變換為Observable<Response<HashMap>>就可以了煌贴。比如：

rx05.png

但現(xiàn)在的需求是，隊列上傳文件锥忿，也就是說牛郑，必須一個任務(wù)完成(成功｜失敗)后才能進行下一個任務(wù)。這樣用flatMap就不可以了敬鬓。(其實后來我考慮過這個問題淹朋，線程的調(diào)度本質(zhì)還是由我聲明出來的線程池來決定的，如果用Schedulers.newThread()钉答，那就會創(chuàng)建多個子線程础芍。但如果用Schedulers.from(Executors.newSingleThreadExecutor())呢？)

需求總是多變的数尿，好在有rxjava可以隨意變換仑性。來吧，我們看看不用單個線程池右蹦，如何實現(xiàn)隊列诊杆。

不能隨意套路歼捐，坑的是自己

之前學(xué)習(xí)rxjava時，看過很多在android中高度使用rxjava的文章刽辙。有一個操作符很有意思-> concat()

The Concat operator concatenates the output of multiple Observables so that they act like a single Observable, with all of the items emitted by the first Observable being emitted before any of the items emitted by the second Observable (and so forth, if there are more than two).

即將多個Observables串起成一個Observable窥岩，直到一個執(zhí)行完畢后再執(zhí)行下一個。

我們可以將這個concat()應(yīng)用在讀取緩存還是請求服務(wù)器, 如果緩存有數(shù)據(jù)宰缤，那就不用請求服務(wù)器了颂翼。

Observable<Data> cache;
Observable<Data> server;
Observable.concat(cache, server)
            .first()

這個也可以用在隊列上傳文件場景上咯。but慨灭，concat()是創(chuàng)建型操作符朦乏，再次變換就不能使用了。不過可以用concatMap(),

Returns a new Observable that emits items resulting from applying a function that you supply to each item emitted by the source Observable, where that function returns an Observable, and then emitting the items that result from concatenating those resulting Observables.

直接看代碼吧

rx06.png

這段寫的特別扭氧骤，為什么又要在一個Observable里又創(chuàng)建一個retrofit相關(guān)的Observable？當(dāng)時想的是筹陵，因為要在upload成功后得刪除文件啊。如果把subscribe放到外層去朦佩，那接收到的全是服務(wù)器response，不知道當(dāng)前的response屬于哪一個file upload语稠。所以我就又寫了次變換宋彼。（這里肯定可以優(yōu)化的，寫的太挫）

在concatMap中接收到from()發(fā)射來的一個Observable输涕，變換成Retrofit請求慨畸，當(dāng)Subscriber標記為onCompleted后再去執(zhí)行下一個Observable莱坎。

到這里還沒完，假如無網(wǎng)絡(luò)又或者服務(wù)器異常先口。在第一個Observable就會失敗型奥，此時還需要繼續(xù)請求嗎碉京？很有可能后面的Observable也都不成功。那加個判斷吧谐宙。concat()可以和first()一起用。concatMap()也是可以的。

rx07.png

If you are only interested in the first item emitted by an Observable, or the first item that meets some criteria, you can filter the Observable with the First operator.

如果first() -> return true; 這樣只取到目前的這個Observable垢箕，后續(xù)的不執(zhí)行了兑巾。

也就是說，只有在上傳成功時return false蒋歌，繼續(xù)執(zhí)行下一個Observable。否則就return true停止修档。

覺醒分割線

我想之前肯定是被concat(cache, db, server).first()整懵逼了府框，一心去套，才寫了上面這么二的代碼的迫靖。等等，容我換個姿勢撕予。

rx10.png

看蜈首，對請求結(jié)果map變換一次就可以啦欠母，如果成功刪除相關(guān)文件，不成功就是個異常了赏淌。Observable.error()踩寇。這樣就跳出了concatMap，也就是說六水，當(dāng)異常發(fā)生時會停止后續(xù)的文件上傳。這樣first()也不需要啦睛榄。除非還有其他額外的停止flag要判斷想帅。

到這里整個workflow就被rxjava梳理完畢了。是不是很有趣？我們來看下代碼全過程咧欣。

rx08.png

還剩最后一個問題：線程調(diào)度轨帜。

之前一直都沒寫線程調(diào)度的地方。subscribeOn放在哪里比較好哮兰？

需求是：在主線程listFile拿到目錄下的所有文件梢什，然后在子線程一個個隊列上傳文件，執(zhí)行完畢后再切換到主線程彈dialog告知結(jié)果嗡午。

這樣來說，每個文件上傳時不需要切換線程狸演，所以調(diào)用retrofit的地方是不需要subscribeOn僻他。如果執(zhí)意要在uploadTrip()后加上subscribeOn(io)，也不是不可以吨拗。只是每個上傳task都在一個新的線程里執(zhí)行的。但實際上哨鸭，我們的文件上傳是個隊列娇妓，完全可以一直在同一個線程里執(zhí)行。所以我在了flatMap后observeOn(io)哈恰。最終執(zhí)行的log如下圖

rx09.png

之前亂嘗試，寫了兩個subscribeOn()着绷，雖然邏輯是對的，但思想上來說是observeOn多次夸楣，subscribeOn一次，幸好有位朋友提醒石洗，才改成了上圖。

好了讲衫，混亂的workflow總算擼成串了孵班。平時看相關(guān)文章總覺得很簡單，無非就是幾個操作符拼接在一起篙程，做了線程切換虱饿。不好理解的就是鏈式思維的轉(zhuǎn)換還有一些操作符：compose transformer等。等到真正應(yīng)用到項目場景中氮发，著實折騰了不少。比如不用flatMap仇祭，改為concatMap颈畸。比如線程調(diào)度。比如放棄使用retrofit+rxjava套路眯娱，重新認識reactive等。

總得來說，當(dāng)理解了rxjava的鏈式思維并對一些復(fù)雜的邏輯重構(gòu)之后贰谣，還是會愛上的。

參考閱讀

1. 理解操作符百宇，還是看官網(wǎng)最佳ReactiveX
2. 感謝小鄧子幫忙梳理流程秘豹。