在Android中使用RxJava

RxJava框架已经出来很久，但是我却是最近才去了解这个框架，还有很多内容没有搞清楚,而且我对这个框架的使用也是更多的停留在异步操作和一些响应式。但是我还是希望能和大家分享下我的学习心得。

RxJava使用的是一个观察者模式。Observable是一个被观察者，Observer是一个观察者，两者间通过subscribe（）进行订阅。

RxJava中的观察者模式

代码表示：

代码

Observable对象被创建时，会将OnSubscribe对象保存。

在观察者Observer是一个接口Interface，它有一个抽象类Subscriber。Subscriber对Observer接口进行了一些扩展，但他们的基本使用方式是完全一致的。

Subscriber相比于Observer多了onStart()方法，这个方法是在事件发送前调用的。还有一个unSubscribe（）是用来取消订阅的。

其实在subscribe（）里面就是以Subscriber进行传递的。而且观察者是一个Observer对象的话，也会转变成Subscriber对象的。

在源码中的实现（不是源码，只是类似实现这样的功能的伪代码）

核心代码

大家可以看到在onSubscribe的call（）是在订阅后才进行回调的，而且OnSubscribe中call（）的传入的Subscriber就是我们所构建的观察者subscriber对象。而在这之前是先执行subscriber的onStart（）方法。所以我们一般可以在onStart（）方法里面做一些准备工作。

create（）方法是RxJava最基本的创造事件序列的方法。基于这个方法，RxJava还提供了一些方法来快捷创建事件队列：

      just()会将传入的对象组一次发送：如：just("1","2","3")就类似于subscriber.onNext("1");subscriber.onNext("2");subscriber("3");subscriber.onCompleted();

     from(T[])/from(Iterale)将传入的数组或Iterable拆分成具体对象后，依次发送出来。(和上面大同小异)

其实这两个方法最终都是会转成create（）来进行发送，而just中的参数大于1个对象时会转变成form（T[]）形式进行发送。而form（T[]）的数组等于1是又会转成just（）进行发送。

在数组长度为1的时候会just方法中会转变成create（），发送数组的过程为：

而在数组长度大于1时，同样会转变成create（），发送数组的过程为：

变换：

如果有这样的逻辑，传入一个url，然后下载图片，再显示在imageview中。其实就涉及都一个url转成bitmap的过程。

在RxJava中提供了一些转变的方法：

map（）可以将OnSubscribe发送过来的事件依次转换成自己想要的类型然后发送给subscriber对象中。

map（）方法中的传递的参数是一个Func1<T,R>对象,T为onSubscribe中发送事件的类型，R为要转变成的类型。在这个对象中有一个回调方法call，这个方法中的参数为onSubscribe中的发送的事件，返回值为要转变的类型。最后这个转变后的对象会被发送到Subscriber的onNext（）中。

map()实现代码

map（）方法只能将事件一个一个的转换，但是有这样的情，比如况我们发送一组文件夹的对象，一个文件夹中又包含有一个图片集合。我们要讲图片一一提取。如果用map（）的话：

虽然能够实现，但是总感觉不是特别方便，因为我们要在Subscriber中做遍历。

flatMap()，RxJava中另一个变换的方法。在flatMap中的参数同样是Func1对象，但是它的泛型结构是这样的Func1<T,Observable<R>>,T为OnSubscriber中发送的事件类型，而R为转变后的类型。

使用flatmap（）

其实在RxJava中的一些列的转变，都是基于lift（）的。

lift（）使用

在lift（）会返回一个新创建的Observable对象

lift（）源码

大家可以看到在lift（）会重新创建一个Observable对象返回，并且会创建一个OnSubscribeLift对象，OnSubscribeLift类是OnSubscribe的一个实现类，会作为新的Observable的onSubscribe对象保存起来。在构造OnSubscribeLift的时候会将当前OnServable对象作为参数（即create（）中的OnSubscribe），而lift（）中的参数--Operator对象也会作为参数传入。

而在OnSubscribe的Call（）的源码：

可以看到，其实在call（）中回调了上一个Observable中的OnSubscribe的call方法。

因为lift会创建一个新的Observable并返回，所以Observale.create().lift().subscribe(Subscriber)的过程可以看成是：

所以最后的订阅操作其实是newObservable与Subscriber对象的订阅。

大致调用流程为：

线程控制

在不指定线程的情况下，以上的所有操作都默认是在同一线程进行的。就是在哪个线程进行订阅操作的话就默认在这个线程执行所有的发送事件和消费事件。

但是RxJava最重要的就是异步操作，是如何实现的呢，就需要用到Scheduler -- 调度器。

在RxJava中内置了几个Scheduler:

我们可以使用subscribeOn()和ObserveOn()两个方法来进行线程控制。

subscriberOn()指定的是事件生产线程：即为OnSubscribe中的call（）运行的线程。

observeOn()指定的是事件消费线程：即为Subscriber所运行的线程。

这次分享就先跟大家分享到这里，如果有哪些错误的地方也希望大家能指出来。因为线程控制的一些原理还不是特别懂，也不敢跟大家瞎分享，所以就写了这么一点。如果大家想了解多一点的话，大家可以看下扔物线的一篇博客《给 Android 开发者的 RxJava 详解》http://gank.io/post/560e15be2dca930e00da1083，个人感觉写得非常好，我的很多分享都是从那里看来的。还有大家在学习的时候可以结合源代码，这样对原理的理解会更加深刻，同时也方便大家理解。

谢谢！！！！