Rxjava实践-把混乱的WORKFLOW撸成串吧

摘要：来吧，我们看看不用单个线程池，如何实现队列。也就是说，只有在上传成功时，继续执行下一个。最终执行的如下图好了，混乱的总算撸成串了。

前几天和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