rxjava

2015-10-14 13:47:06   25  举报





AI智能生成

RxJava是一个使用可观察序列来编写异步和基于事件的程序的库。它通过使用Observables和Subscribers来实现响应式编程，使得代码更加简洁、易读、易维护。RxJava提供了丰富的操作符，如map、filter、reduce等，可以方便地对数据进行处理和转换。此外，RxJava还支持多线程执行，可以充分利用CPU资源，提高程序的性能。总之，RxJava是一个非常强大的工具，可以帮助开发者更好地处理异步任务和事件驱动的程序。

作者其他创作

大纲/内容

与常规订阅模式的区别

RxJava 的事件回调方法除了普通事件 onNext() （相当于 onClick() / onEvent()）之外，还定义了两个特殊的事件：onCompleted() 和 onError()。【使得能干涉到整个任务队列】

基本实现

创建 Observer

Observer

与Subscriber 基本使用方式一样，实质上，在 RxJava 的 subscribe 过程中，Observer 也总是会先被转换成一个 Subscriber 再使用。所以如果你只想使用基本功能，选择 Observer 和 Subscriber 是完全一样的。

Subscriber

相对于Observer新增了onStart()方法，它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。这是一个可选方法，默认情况下它的实现为空。需要注意的是，如果对准备工作的线程有要求（例如弹出一个显示进度的对话框，这必须在主线程执行）， onStart() 就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法

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

创建 Observable

Observable

Observable 即被观察者，它决定什么时候触发事件以及触发怎样的事件。 RxJava 使用 create() 方法来创建一个 Observable ，并为它定义事件触发规则

传入了一个 OnSubscribe 对象作为参数。OnSubscribe 会被存储在返回的 Observable 对象中，它的作用相当于一个计划表，当 Observable 被订阅的时候，OnSubscribe 的 call() 方法会自动被调用，事件序列就会依照设定依次触发（对于上面的代码，就是观察者Subscriber 将会被调用三次 onNext() 和一次 onCompleted()）。这样，由被观察者调用了观察者的回调方法，就实现了由被观察者向观察者的事件传递，即观察者模式。

接口

subscriber()方法

调用 Subscriber.onStart() 。这个方法在前面已经介绍过，是一个可选的准备方法

调用 Observable 中的 OnSubscribe.call(Subscriber) 。在这里，事件发送的逻辑开始运行。从这也可以看出，在 RxJava 中， Observable 并不是在创建的时候就立即开始发送事件，而是在它被订阅的时候，即当 subscribe() 方法执行的时候

将传入的 Subscriber 作为 Subscription 返回。这是为了方便 unsubscribe().

subscribeOn

指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。

observeOn()

指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。

Subscribe (订阅)

创建了 Observable 和 Observer 之后，再用 subscribe() 方法将它们联结起来，整条链子就可以工作了。

采用Observable.subscribe(Subscriber)这种略蹩脚语义的api是出于流式api的考虑，因为之后的操作都是针对Observable

Scheduler

在 RxJava 的默认规则中，事件的发出和消费都是在同一个线程的。也就是说，如果只用上面的方法，实现出来的只是一个同步的观察者模式。观察者模式本身的目的就是『后台处理，前台回调』的异步机制，因此异步对于 RxJava 是至关重要的。而要实现异步，则需要用到 RxJava 的另一个概念： Scheduler

线程控制

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

接口

内置了几个 Scheduler

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 主线程运行

概念

Observable （被/可观察者）

通过 subscribe() 方法实现和Observer 的订阅关系

在需要的时候发出事件来通知 Observer

Observer （观察者）

subscribe （订阅）

event（事件）

Scheduler

异步对于 RxJava 是至关重要的。而要实现异步，则需要用到Scheduler

变换

map() 方法将参数中的 String 对象转换成一个 Bitmap 对象后返回，而在经过 map() 方法后，事件的参数类型也由 String 转为了 Bitmap。这种直接变换对象并返回的，是最常见的也最容易理解的变换。不过 RxJava 的变换远不止这样，它不仅可以针对事件对象，还可以针对整个事件队列，这使得 RxJava 变得非常灵活。