开源库之RxJava & RxAndroid基本使用

Andrid Studio中引用RxAndroid

compile 'io.reactivex:rxandroid:1.2.0'

Hello World

RxJava最核心的两个东西是Observables（被观察者，事件源）和Subscribers（观察者）。Observables发出一系列事件，Subscribers处理这些事件（例如：触摸事件，web接口调用返回的数据...）。一个Observable可以发出零个或者多个事件，直到结束或者出错。每发出一个事件，就会调用它的Subscriber的onNext方法，最后调用Subscriber.onError()或者Subscriber.onError()结束。
Rxjava的看起来很想设计模式中的观察者模式，但是有一点明显不同，那就是如果一个Observerble没有任何的的Subscriber，那么这个Observable是不会发出任何事件的。
创建一个Observable对象很简单，直接调用Observable.create即可。

Observable<String> myObservable = Observable.create(//创建事件源
        new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> sub) {
                sub.onNext("Hello, world!");
                sub.onCompleted();//该方法必须主动调用
            }
        }
);

这里定义的Observable对象myObservable仅仅发出一个"Hello World"字符串，然后就结束了。

接着我们创建一个Subscriber来处理Observable对象发出的字符串。

Subscriber<String> mySubscriber = new Subscriber<String>() {//创建观察者
    @Override
    public void onNext(String s) {
        lg.e("RxJava 测试结果：" + s);
    }

    @Override
    public void onCompleted() {
        lg.e("RxJava onCompleted");
    }

    @Override
    public void onError(Throwable e) {
        lg.e("RxJava onError");
    }
};

这里subscriber仅仅就是打印observable发出的字符串。

Subscriber实现了Observer接口，而在 RxJava 的 subscribe 过程中，Observer 也总是会先被转换成一个 Subscriber 。所以如果你只想使用基本功能，选择 Observer 和 Subscriber 是完全一样的。它们的区别对于使用者来说主要有两点：

1.onStart(): 这是 Subscriber 增加的方法。它会在 subscribe 刚开始而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置，默认空实现。需要注意的是，onStart() 总是在 subscribe 所发生的线程被调用，并不能保证在主线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法，具体可以在后面的文中看到。

2.unsubscribe(): 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。调用该方法将使得Subscriber 不再接收事件。一般在调用前使用 isUnsubscribed() 先判断一下状态。 unsubscribe()方法很重要，因为在 subscribe() 之后， Observable 会持有 Subscriber 的引用，这个引用如果不能及时被释放，将有内存泄露的风险。所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop() 等方法中）调用 unsubscribe() 来解除引用关系，以避免内存泄露的发生。

通过subscribe函数就可以将我们定义的myObservable对象和mySubscriber对象关联起来，这样就完成了subscriber对observable的订阅。

myObservable.subscribe(mySubscriber);

一旦mySubscriber订阅了myObservable，myObservable就是调用mySubscriber对象的onNext和onComplete|onError方法。

变换-> Map &flatMap操作符

RxJava 提供了对事件序列进行变换的支持，这是它的核心功能之一，也是大多数人说『RxJava 真是太好用了』的最大原因。所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。

使用场景：当需要修改接收的内容时，

a.可以在Observable中直接修改字符串,局限在于

1.若Observable是第三方提供的，可能不允许更改

2.此时修改属于批量修改，凡是订阅此Observable的数据均会改变

b.在订阅者端处理：

1.与希望Subscribers越轻量越好的原则冲突

2.根据响应式函数编程的概念，Subscribers更应该做的事情是"响应"，响应Observable发出的事件，而不是去修改

基于map针对对象进行变换

线程控制 -> Scheduler

在不指定线程的情况下， RxJava 遵循的是线程不变的原则，即：在调用 subscribe()的线程生产事件；在生产事件的线程消费事件。如果需要切换线程，就需要用到 Scheduler（调度器）。

1) Scheduler 的 API (一)

在RxJava 中，Scheduler ——调度器，相当于线程控制器，RxJava 通过它来指定每一段代码应该运行在什么样的线程。RxJava 已经内置了几个 Scheduler ，它们已经适合大多数的使用场景：

a.Schedulers.immediate(): 直接在当前线程运行（与），相当于不指定线程。这是默认的 Scheduler。

b.Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。

c.Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread()更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。

d.Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。

另外， Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。

接下来就可以使用subscribeOn() 和 observeOn() 两个方法对线程进行控制了。

subscribeOn(): 即事件产生的线程。

observeOn(): 指定Subscriber所在的线程(map、subscribe等方法的执行域)，即事件消费的线程。

示例如下：

Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {//这里所属线程受subscribeOn控制
        lg.e("isFunctionInMainThread-call:" + UtilsThread.isFunctionInMainThread());
        subscriber.onNext("***");
        subscriber.onCompleted();
    }
}).subscribeOn(Schedulers.io()).map(new Func1<String, String>() {//该map操作因为没调用observeOn指定线程，默认使用Observable创建所属线程(即create方法所属)
    @Override
    public String call(String s) {
        lg.e("isFunctionInMainThread01:" + UtilsThread.isFunctionInMainThread());
        return s;
    }
}).observeOn(Schedulers.io()).map(new Func1<String, String>() {
    @Override
    public String call(String s) {
        lg.e("isFunctionInMainThread02:" + UtilsThread.isFunctionInMainThread());
        return s;
    }
}).observeOn(AndroidSchedulers.mainThread()).map(new Func1<String, String>() {//这里的map操作在主线程
    @Override
    public String call(String s) {
        lg.e("isFunctionInMainThread03:" + UtilsThread.isFunctionInMainThread());
        return s;
    }
}).observeOn(Schedulers.io()).subscribe(new Subscriber<String>() {//这里表示subscribe在io线程
    @Override
    public void onCompleted() {
        lg.e("isFunctionInMainThread-subscribe:" + UtilsThread.isFunctionInMainThread());
    }

    @Override
    public void onError(Throwable e) {
        lg.e("onError");
    }

    @Override
    public void onNext(String s) {
        lg.e("onNext with " + s, "isFunctionInMainThread-subscribe:" + UtilsThread.isFunctionInMainThread());
    }
});

运行结果如下所示，

思考：

flatmap[铺平]相关介绍

来自为知笔记(Wiz)