Android 中的 rxjava

2016/3/20 posted in Android

介绍 rxjava之前要先明白一个前置概念 ReactiveX

ReactiveX定义

ReactiveX is an API that focuses on asynchronous composition and manipulation of observable streams of data or events by using a combination of the Observer pattern, Iterator pattern, and features of Functional Programming. Handling real-time data is a common occurrence, and having an efficient, clean, and extensible approach to handling these scenarios is important. Using Observables and operators to manipulate them, ReactiveX offers a composable and flexible API to create and act on streams of data while simplifying the normal concerns of asynchronous programming like thread creation and concurrency issues.

http://reactivex.io/intro.html

ReactiveX是Reactive Extensions的缩写，一般简写为Rx

ReactiveX 是一个使用可观察数据流进行异步编程的编程接口，ReactiveX结合了观察者模式、迭代器模式和函数式编程的精华

RX模式

使用观察者模式

创建：Rx可以方便的创建事件流和数据流
组合：Rx使用查询式的操作符组合和变换数据流
监听：Rx可以订阅任何可观察的数据流并执行操作

简化代码

函数式风格：对可观察数据流使用无副作用的输入输出函数，避免了程序里错综复杂的状态
简化代码：Rx的操作符通通常可以将复杂的难题简化为很少的几行代码
异步错误处理：传统的try/catch没办法处理异步计算，Rx提供了合适的错误处理机制
轻松使用并发：Rx的Observables和Schedulers让开发者可以摆脱底层的线程同步和各种并发问题
使用Observable的优势

Rx扩展了观察者模式用于支持数据和事件序列，添加了一些操作符，它让你可以声明式的组合这些序列，而无需关注底层的实现：如线程、同步、线程安全、并发数据结构和非阻塞IO。

响应式编程

Rx提供了一系列的操作符，你可以使用它们来过滤(filter)、选择(select)、变换(transform)、结合(combine)和组合(compose)多个Observable，这些操作符让执行和复合变得非常高效。

以上对RX 介绍来自RX官方的中文文档，比较抽象，可以先看具体代码，回头再看定义会比较容易理解

Rxjava

RxJava是目前在Android开发者中新兴热门的函数库。

基本概念

观察者模式不细说了，详见 http://xuyushi.github.io/2016/02/20/观察者模式/

响应式代码的基本组成部分是Observables和Subscribers（事实上Observer才是最小的构建块，但实践中使用最多的是Subscriber，因为Subscriber才是和Observables的对应的。）。Observable发送消息，而Subscriber则用于消费消息。

消息的发送是有固定模式的。Observable可以发送任意数量的消息（包括空消息），当消息被成功处理或者出错时，流程结束。Observable会调用它的每个Subscriber的Subscriber.onNext()函数，并最终以Subscriber.onComplete()或者Subscriber.onError()结束。

这看起来像标准的观察者模式，但不同的一个关键点是：Observables一般只有等到有Subscriber订阅它，才会开始发送消息（术语上讲就是热启动Observable和冷启动Observable。热启动Observable任何时候都会发送消息，即使没有任何观察者监听它。冷启动Observable只有在至少有一个订阅者的时候才会发送消息。换句话说，如果没有订阅者观察它，那么将不会起什么作用。

Hello,World！

喜闻乐见的 hello world 时间~

gradle 加入

compile 'io.reactivex:rxandroid:1.1.0'
compile 'io.reactivex:rxjava:x.y.z'

创建一个基本的Observable，被观察的对象，事件的发起者

Observable<String> myObservable = Observable.create(
                new Observable.OnSubscribe<String>() {
                    @Override
                    public void call(Subscriber<? super String> sub) {
                        sub.onNext("Hello, world!");
                        sub.onCompleted();
                    }
                }
        );

创建事件的观察者，接受事件，onNext是接受到事件的回调，onCompleted是Observable发送完一系列时间结束的回调，onError是出错的回调

   Subscriber<String> mySubscriber = new Subscriber<String>() {
       @Override
       public void onNext(String s) {
           Log.d(TAG, "onNext: " + s);
       }

       @Override
       public void onCompleted() {
           Log.d(TAG, "onCompleted: ");
       }

       @Override
       public void onError(Throwable e) {
           Log.d(TAG, "onError: " + e.toString());
       }
   };

此时是不会打印出 log 的，原因是我们没有将mySubscriber订阅到Observable中

myObservable.subscribe(mySubscriber);

LOG

D/MainActivity: onNext: Hello, world!
D/MainActivity: onCompleted:

更简洁的代码

RxJava为常见任务提供了很多内建的Observable创建函数。在以下这个例子中，Observable.just()发送一个消息然后完成，功能类似上面的代码

Observable<String> myObservable =
    Observable.just("Hello, world!");

在Subscriber中，我们并不需要关心onCompleted和onError
的情况，我们可以简化代码

Action1<String> onNextAction = new Action1<String>() {
    @Override
    public void call(String s) {
        Log.d(TAG, "onNext: " + s);
    }
};

我们来看看 Action1是什么鬼

/**
 * A one-argument action.
 */
public interface Action1<T> extends Action {
    void call(T t);
}

Action1 是一个接口，里面只有一个带泛华参数的call方法
这个方法是无参无返回值的；由于onCompleted() 方法也是无参无返回值的，因此 Action1 可以被当成一个包装对象，将 onCompleted() 的内容打包起来将自己作为一个参数传入 subscribe() 以实现不完整定义的回调。

同理 onError 和onComplete

Action1<String> onNextAction = new Action1<String>() {
    // onNext()
    @Override
    public void call(String s) {
        Log.d(tag, s);
    }
};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
    // onError()
    @Override
    public void call(Throwable throwable) {
        // Error handling
    }
};
Action0 onCompletedAction = new Action0() {
    // onCompleted()
    @Override
    public void call() {
        Log.d(tag, "completed");
    }
};

// 自动创建 Subscriber ，并使用 onNextAction 来定义 onNext()
observable.subscribe(onNextAction);
// 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);
// 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);

ActionX X 表示参数的个数，可以为0

线程控制

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

我们可以随意控制 Observable 和Subscriber执行的线程

subscribeOn(): 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。
Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。
Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。
Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和 newThread() 差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
Schedulers.computation(): 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。

变换

比如我想在 hello world 中，添加一些字符串，我们可以在 Observable 发送的时候添加

Observable.just("Hello, world! add someting.")
    .subscribe(s -> Log,d("TAG",s));

或者在接收的时候添加

Observable.just("Hello, world!")
    .subscribe(s -> Log,d("TAG",s + "add someting."));

这里使用了 lambda表达式，使得语句更精简,Android N 即将开放对lambda表达式表达式的支持，目前 IDEA 已经支持了 lambda

但是这样代码就失去了通用性

Operators

Operators在消息发送者Observable和消息消费者Subscriber之间起到操纵消息的作用。RxJava拥有大量的opetators

map

map() operator可以被用于将已被发送的消息转换成另外一种形式

Observable.just("Hello, world!")
    .map(new Func1<String, String>() {
        @Override
        public String call(String s) {
            return s + "add someting.";
        }
    })
    .subscribe(s -> Log.d("TAG",s));

扩展

Observable和Subscriber能完成任何事情。

Observable可以是一个数据库查询，Subscriber获得查询结果然后将其显示在屏幕上。你的Observable可以是屏幕上的一个点击，Subscriber响应该事件。你的Observable可以从网络上读取一个字节流，Subscriber将其写入本地磁盘中。这是一个可以处理任何事情的通用框架。

Observable和Subscriber与它们之间的一系列转换步骤是相互独立的。

我们可以在消息发送者Observable和消息消费者Subscriber之间加入任意多个想要的map()函数。这个系统是高度可组合的：它很容易对数据进行操纵。只要operators符合输入输出的数据类型，那么我可以得到一个无穷尽的调用链。

flatMap

假如我们现在有这样一个方法
Observable<List<String>> query(String text);

我们希望打印出这些 string

 query("Hello, world!")
    .subscribe(urls -> {
        for (String url : urls) {
            System.out.println(url);
        }
    });

毫无疑问这样的代码是丑陋的，我们失去了对数据流进行变幻的机会

我们也可以在事件分发钱对 String list 做拆分，比如

query("Hello, world!")
    .subscribe(urls -> {
        Observable.from(urls)
            .subscribe(url -> System.out.println(url));
    });

迷之缩进又出来了，我们更希望看到的应该是链式的调用的关系

我们可以使用Observable.flatMap() 接受一个Observable然后分发成其他的Observable

query("Hello, world!")
    .flatMap(new Func1<List<String>, Observable<String>>() {
        @Override
        public Observable<String> call(List<String> urls) {
            return Observable.from(urls);
        }
    })
    .subscribe(url -> System.out.println(url));
 ```
 
### 更多的Operators
 Operators 可以将负责的逻辑分解为多段的链式调用，`Subscriber`并不需要知道这负责的链式关系，只管接受处理最后转换完成的事件
 
 https://github.com/ReactiveX/RxJava/wiki/Alphabetical-List-of-Observable-Operators

### 结束处理

`onComplete()` 和 `onError()`表示分发事件的结束，两者是互斥的，在`Subscriber`可是监听是`onComplete()` 和 `onError()`

* `onComplete()`：事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。
* 


```java
Observable.just("Hello, world!")
    .map(s -> potentialException(s))
    .map(s -> anotherPotentialException(s))
    .subscribe(new Subscriber<String>() {
        @Override
        public void onNext(String s) { System.out.println(s); }

        @Override
        public void onCompleted() { System.out.println("Completed!"); }

        @Override
        public void onError(Throwable e) { System.out.println("Ouch!"); }
    });

当Exception被抛出时，onError()就会被调用
在operators不需要处理Exception，只需要在Subscriber中处理就可以了

Subscriptions

当我们调用Observable.subscribe()时，会返回Subscription，它表示Observable 和 Subscriber 的链接

Subscription subscription = Observable.just("Hello, World!")
    .subscribe(s -> System.out.println(s));

可以通过 Subscription来断开链接

subscription.unsubscribe();

Android 中的 rxjava

参考

http://www.captechconsulting.com/blogs/getting-started-with-rxjava-and-android
http://reactivex.io/intro.html
https://asce1885.gitbooks.io/android-rd-senior-advanced/content/che_di_le_jie_rxjava_ff08_yi_ff09_ji_chu_zhi_shi.html
http://gank.io/post/560e15be2dca930e00da1083
http://blog.danlew.net/2014/09/22/grokking-rxjava-part-2/
http://blog.alexwan1989.com/2016/02/23/RxJava读书笔记(一)/

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

传统的嵌套请求需要使用嵌套的 Callback 来实现。而通过 flatMap() ，可以把嵌套的请求写在一条链中，从而保持程序逻辑的清晰。

defer()中的代码直到被订阅才会执行。

1、just()
指定特定的对象
2、repeat()
指定重复的次数
3、defer()
指定订阅的时机
4、range()
指定指定起始点和个数
5、interval()
指定间隔时间和对应的时间单位
6、timer()
指定延迟执行
五、过滤Observer
1、filter()
过滤我们观测序列中不想要的值
2、take()
过滤指定前几个返回值
3、takeLast()
过滤指定后几个返回值，仅可用于完成的序列。
4、distinct()
去掉重复值
5、distinctUntilChanged()
执行去重，直到值发生变化时。
6、first()
获取第一个元素
7、last()
获取最后一个元素
8、skip()
与take()相反：跳过前几个元素
9、skipLast()
与takeLast()相反：跳过后几个元素
10、elementAt()
返回指定位置的元素
11、elementAtOrDefault()
返回指定位置元素，不存在时返回默认
12、sample()
指定的时间间隔里由Observable返回最近一次的数值：指定时间间隔，时间单位
13、throttleFirst()
指定的时间间隔里由Observable返回第一个的数值：指定时间间隔，时间单位
14、timeout()
设定的时间间隔内如果没有得到一个值则返回一个错误，超时获取到的元素不会被发送。
15、Debounce()
过滤掉由Observable发送的速率过快的元素，如果在一个指定的时间间隔过去了仍旧没有发射一个，那么它将发射最后的那个。
六、转换Observables
1、*map
• map()
• flatMap()
• concatMap()
• flatMapIterable()
• switchMap()
• scan()
（1）、map()
指定Func对元素进行操作
（2）、flatMap()
不能保证最终的顺序与源值顺序一致，在处理大量的Observable时，任意的Observable发生错误时，flatMap都会调用自己的onError()，并放弃整条链。
（3）、cancatMap()
解决flatMap()交叉问题，提供能够将发送的元素连续flat的函数，而不是合并。
（4）、flatMapIterable()
将源数据两两结合生成Iterable
（5）、switchMap()
每当源Observable发射一个新的数据项（Observable）时，它将取消订阅并停止监视之前那个数据项产生的Observable
（6）、scan()
指定函数来累加计算最终结果值，类似累加器。
2、groupby()
从列表中按照指定的规则：groupBy()来分组元素
3、buffer()
将源Observable变换一个新的Observable，新的Observable为指定count数的列表。指定count数时，还可带一个skip参数指定skip后返回count个的一个列表。同时还有一个timespan参数，指定间隔多少时间，来发送指定的列表。
4、window()
与buffer()相似，与buffer不同的是，它发送的是Observale对象。window(skip = 3 , count = 2).Observables中的每一个都发射原始Observable数据的一个子集，数量由count指定,最后发射一个onCompleted()结束.正如buffer()一样,window()也有一个skip参数。
5、cast()
源Observable中的每一项数据都转换为新的类型，把它变成了不同的类型

« Material Design

单身汪的自白 »

Android 中的 rxjava

ReactiveX定义

RX模式

使用观察者模式

简化代码

响应式编程

Rxjava

基本概念

Hello,World！

更简洁的代码

线程控制

变换

Operators

map

扩展

flatMap

Subscriptions

Android 中的 rxjava

参考

Xu Yushi's Blog

Categories

Recent Posts