摘要：來吧，我們看看不用單個線程池，如何實現隊列。也就是說，只有在上傳成功時，繼續執行下一個。最終執行的如下圖好了，混亂的總算擼成串了。

前幾天和Stay（有心課堂創始人）討論了一個rxjava場景，覺得很有意思，這里也和大家分享下。通過還原一個真實的開發過程，來感受下rxjava的便利之處。

不認識Stay的可以看看他的博客吧：http://www.jianshu.com/users/...

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

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

看起來好像是沒什么問題，一個for循環搞定。一個task失敗了不影響另一個task。每個task run在一個多帶帶的子線程。

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

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

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

workflow分三步：

locate to data dir

list files under data dir

filter files with .loc suffix

換成rxjava非常容易

先發射一個data目錄路徑

需求是多次上傳文件，得用flatMap將data映射成一個Observable

  2.1 當然你可以選擇直接listFile(filter)，但這樣回調又套回調，不是很好看。
  2.2 用filter操作符將發射來的File[]過濾

比如像2.1這樣寫

或者像2.2這樣寫

注意，在flatMap中又用from()操作符將File[]變換成一個個Observable發射。

假如你的API接口可以接收多個文件，其實也不用這樣寫。直接在flatMap中拼接RequestBody，調用API請求就可以了。比如像下圖這樣寫：
![rx04(http://img.blog.csdn.net/2016...
無奈需求是上傳loc文件同時還會再帶上一個sensor文件，所以就不能像上述這樣寫。

接下來的workflow就很有趣了。現在有了多個Observable，一個個上傳就是了。

如果不考慮隊列，不考慮無網或上傳失敗情況。完全再來一個flatMap將Observable變換為Observable>就可以了。比如：

但現在的需求是，隊列上傳文件，也就是說，必須一個任務完成(成功｜失敗)后才能進行下一個任務。這樣用flatMap就不可以了。(其實后來我考慮過這個問題，線程的調度本質還是由我聲明出來的線程池來決定的，如果用Schedulers.newThread()，那就會創建多個子線程。但如果用Schedulers.from(Executors.newSingleThreadExecutor())呢？)

需求總是多變的，好在有rxjava可以隨意變換。來吧，我們看看不用單個線程池，如何實現隊列。

之前學習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，直到一個執行完畢后再執行下一個。

我們可以將這個concat()應用在讀取緩存還是請求服務器, 如果緩存有數據，那就不用請求服務器了。

Observable cache;
Observable server;
Observable.concat(cache, server)
            .first()

這個也可以用在隊列上傳文件場景上咯。but，concat()是創建型操作符，再次變換就不能使用了。不過可以用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.

直接看代碼吧

這段寫的特別扭，為什么又要在一個Observable里又創建一個retrofit相關的Observable？當時想的是，因為要在upload成功后得刪除文件啊。如果把subscribe放到外層去，那接收到的全是服務器response，不知道當前的response屬于哪一個file upload。所以我就又寫了次變換。（這里肯定可以優化的，寫的太挫）

在concatMap中接收到from()發射來的一個Observable，變換成Retrofit請求，當Subscriber標記為onCompleted后再去執行下一個Observable。

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

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，后續的不執行了。

也就是說，只有在上傳成功時return false，繼續執行下一個Observable。否則就return true停止。

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

看，對請求結果map變換一次就可以啦，如果成功刪除相關文件，不成功就是個異常了。Observable.error()。這樣就跳出了concatMap，也就是說，當異常發生時會停止后續的文件上傳。這樣first()也不需要啦。除非還有其他額外的停止flag要判斷。

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

還剩最后一個問題：線程調度。

之前一直都沒寫線程調度的地方。subscribeOn放在哪里比較好？

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

這樣來說，每個文件上傳時不需要切換線程，所以調用retrofit的地方是不需要subscribeOn。如果執意要在uploadTrip()后加上subscribeOn(io)，也不是不可以。只是每個上傳task都在一個新的線程里執行的。但實際上，我們的文件上傳是個隊列，完全可以一直在同一個線程里執行。所以我放在了flatMap中。最終執行的log如下圖

好了，混亂的workflow總算擼成串了。平時看相關文章總覺得很簡單，無非就是幾個操作符拼接在一起，做了線程切換。不好理解的就是鏈式思維的轉換還有一些操作符：compose transformer等。等到真正應用到項目場景中，著實折騰了不少。比如不用flatMap，改為concatMap。比如線程調度。比如放棄使用retrofit+rxjava套路，重新認識reactive等。

總得來說，當理解了rxjava的鏈式思維并對一些復雜的邏輯重構之后，還是會愛上的。

參考閱讀

理解操作符，還是看官網最佳ReactiveX

感謝小鄧子幫忙梳理流程。

有心課堂，傳遞給你的不僅僅是技術！如果你有任何問題，請加入我們的QQ討論群：172448270