RxJava for Android学习笔记

RxJava "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库）。

Rxjava 来说可以用三个词来概括　“异步”　“高效”　“灵活”

Android工程引入RxJava-android 请引入Rxandroid库： compile 'io.reactivex:rxandroid:1.2.1' 一简介 1.1 Rx结构 响应式编程的主要组成部分是observable, operator和susbscriber,一般响应式编程的信息流如下所示： Observable -> Operator 1 -> Operator 2 -> Operator 3 -> Subscriber 也就是说，observable是事件的生产者，subscriber是事件最终的消费者。 因为subscriber通常在主线程中执行，因此设计上要求其代码尽可能简单，只对事件进行响应，而修改事件的工作全部由operator执行。 在响应式编程中，应该牢记以下两点：everything is a stream（一切皆流）don't break the chain（不要打断链式结构） 1.2 最简单的模式 如果我们不需要修改事件，就不需要在observable和subscriber中插入operator。这时的Rx结构如下： Obsevable -> Subscriber 这看起来很像设计模式中的观察者模式，他们最重要的区别之一在于在没有subscriber之前，observable不会产生事件。

拓展的观察者模式

二. 基本实现

1) 创建 Observer Observer 即观察者，它决定事件触发的时候将有怎样的行为。 2) 创建 Observable Observable 即被观察者，它决定什么时候触发事件以及触发怎样的事件。 3) Subscribe (订阅) 创建了 Observable和 Observer之后，再用 subscribe()方法将它们联结起来，整条链子就可以工作了。

observable.subscribe(observer);// 或者：observable.subscribe(subscriber);

整个过程中对象间的关系

三. 线程控制 线程控制 —— Scheduler (一) 在RxJava 中，Scheduler ——调度器，相当于线程控制器，RxJava 通过它来指定每一段代码应该运行在什么样的线程。RxJava 已经内置了几个 Scheduler，它们已经适合大多数的使用场景： Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。 Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。 Schedulers.io(): I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。 Schedulers.computation(): 计算所使用的 Scheduler。 Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。 示例：

Observable.just(1, 2, 3, 4)
 .subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
 .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
 .subscribe(new Action1<Integer>() { 
@Override public void call(Integer number) { 
Log.d(tag, "number:" + number); } });

线程控制：Scheduler (二) 线程的自由控制:可以利用 subscribeOn()结合 observeOn()来实现线程控制，让事件的产生和消费发生在不同的线程。

Observable.just(1, 2, 3, 4) // IO 线程，由 subscribeOn() 指定 .subscribeOn(Schedulers.io()) 
.observeOn(Schedulers.newThread()) 
.map(mapOperator) // 新线程，由 observeOn() 指定
 .observeOn(Schedulers.io())
 .map(mapOperator2) // IO 线程，由 observeOn() 指定 .observeOn(AndroidSchedulers.mainThread) 
.subscribe(subscriber); // Android 主线程，由 observeOn() 指定

多个 subscribeOn()和 observeOn() 混合使用时，线程调度状态

四.操作符 RxJava提供了很多便捷的操作符，即上文提到的Operator,常用到的Operator例如map,flatmap等 map( ) — 对序列的每一项都应用一个函数来变换Observable发射的数据序列

Observable.just("images/logo.png") // 输入类型 String .map(new Func1<String, Bitmap>() { 
@Override public Bitmap call(String filePath) { 
// 参数类型 String
 return getBitmapFromPath(filePath); // 返回类型 Bitmap } }) .subscribe(new Action1<Bitmap>() { 
@Override public void call(Bitmap bitmap) { 
// 参数类型 Bitmap showBitmap(bitmap); } });

map()方法将参数中的 String对象转换成一个 Bitmap对象后返回，而在经过 map()方法后，事件的参数类型也由 String转为了 Bitmap。这种直接变换对象并返回的，是最常见的也最容易理解的变换。

flatMap( ), concatMap( ), and flatMapIterable( ) — 将Observable发射的数据集合变换为Observables集合，然后将这些Observable发射的数据平坦化的放进一个单独的Observable

Student[] students = ...;
Subscriber<Course> subscriber = new Subscriber<Course>() { 
@Override public void onNext(Course course) { 
Log.d(tag, course.getName()); } ...};
Observable.from(students) 
.flatMap(new Func1<Student, Observable<Course>>() { 
@Override public Observable<Course> call(Student student) { 
return Observable.from(student.getCourses()); } }) 
.subscribe(subscriber);

从上面的代码可以看出， flatMap()和 map()有一个相同点：它也是把传入的参数转化之后返回另一个对象。但需要注意，和 map()不同的是， flatMap()中返回的是个 Observable对象，并且这个 Observable对象并不是被直接发送到了 Subscriber的回调方法中。 五.java回调地狱 如果你是个Callback高手，善用并且能够用好FutureTask。 那么在Android中你的代码可能会频繁的使用async+callbacks，或者service composition+ error handing，developer productivity 。 那么关于异步回调的逻辑，你会写成这样getData(Callback)、这样Future<T> getData()，还是这样Future<List<T>> getData()，甚至这样Future<List<Future<T>>> getData()　。。。。。。无法表达CallBack给java菜鸟带来的压力了。。。异步回调的过程中必须要求时刻保持清醒的头脑，灵活的意识，剑拔弩张的情绪，否则是写不好回调事件的 搞张图提神醒脑下:

everything is a stream（一切皆流）

然而使用RxJava的操作符，我们可以避免这些烦人甚至糟糕的回调，让结构和思路看起来更清晰，通过组合API，只需要约定最终的结果Observable就行了。 并且scheduler的出现，不仅解放了线程的切换，让UI线程与工作线程间的跳转变得简单，而且，它的API很丰，也提供了很多使用常见的建议，比如，适用计算任务的Schedulers.computation( )；处理密集IO任务的Schedulers.io( )；以及Schedulers.trampoline( )能够有效避免StackOverflowError，所以非常适合函数的递归调用。 五.与EVENTBUS的区别 EventBus是一个发布 / 订阅的事件总线。简单点说，就是两人约定好怎么通信，一人发布消息，另外一个约定好的人立马接收到你发的消息。 主要区别是，rx里面当建立起订阅关系时，你可以用操作符做任何处理（比如转换数据，更改数据等等），而且他能处理异步的操作。 eventbus 就相当于广播，发送了，总能接收到，他在发送后是不能做任何的数据改变，如果要改变，又要重新post一次。 六.RxJava多带来的好处 先来上张最喜欢的图：

Observables 和 Observers

１．解决了异步回调CallBack的问题 ２．解决线程切换问题，封装了各种并发实现，如threads, pools, event loops, fibers, actors。 ３．能够知道什么时候订阅者已经接收了全部的数据。 ４．所有的错误全部在onError中处理，在操作符不需要处理异常 ５．轻量，无依赖库、Jar包小于1M ６．Java中如果不使用观察者模式，数据都是主动获取，即Pull方式，对于列表数据，也是使用Iterator轮询获取。RxJava由于用到了观察者模式，数据是被动获取，由被观察者向观察者发出通知，即Push方式。 ７．数据传递方式同步和异步都是链式调用，即operation1 -> operation2 -> operation3，这种做法的好处就是即时再复杂的逻辑都简单明了，不容易出错。 ８．比观察者模式功能更强大，在onNext()回调方法基础上增加了onCompleted()和OnError()，当事件执行完或执行出错时回调。此外还可以很方便的切换事件生产和消费的线程。事件可以组合处理。 以上是看了这些东西的新的，关于架构方面，结合mvp架构和retrofit等结构化框架流式框架会更清晰