RxJava的观察者模式(二)

上一篇中我们了解了什么是RxJava,用一个词来总结就是异步。
这里我们来讲讲RxJava的异步实现。它是通过一种扩展的观察者模式来实现。

一、观察者模式

先一下观察者模式。
  观察者模式面向的需求是:观察者对被观察者的某种变化作出反应。比如警察抓小偷,警察需要在小偷作案时实施抓捕。在这里面小偷是被观察者,警察是观察者。而程序的观察者模式跟真正的观察略有不同,观察者不需要时时刻刻顶着被观察者,而是采用注册(Register)或者被称为订阅(Subscribe)方式告诉观察者:我需要你的某种状态,你要在你变化的时候通知我。
  Android开发中典型的观察者模式就是监听器事件Listener。对设置OnClickListener来说,View是被观察者,OnClickListener是观察者,两者通过setOnClickListener()方法达成注册(订阅)关系。订阅之后用户点击按钮的瞬间,Android Framework 就会将点击事件发送给已经注册的 OnClickListener。
OnClickListener的模式图


RxJava的观察者模式(二)_第1张图片
Paste_Image.png

如图所示,通过 setOnClickListener()方法,Button持有 OnClickListener的引用(这一过程没有在图上画出);当用户点击时,Button自动调用 OnClickListener的 onClick()方法。另外,如果把这张图中的概念抽象出来(Button -> 被观察者、OnClickListener-> 观察者、setOnClickListener()-> 订阅,onClick() -> 事件),就由专用的观察者模式(例如只用于监听控件点击)转变成了通用的观察者模式。如下图:

RxJava的观察者模式(二)_第2张图片
Paste_Image.png

而 RxJava 作为一个工具库,使用的就是通用形式的观察者模式。

二、RxJava的观察者模式

RxJava的四个基本概念
  Observable(被观察者)
  Observer(观察者)
  subscribe(订阅)
  事件
  Oservable和Observer通过 subscribe()方法实现订阅关系,从而 Observable可以在需要的时候发出事件来通知 Observer。
  与传统观察者模式不同, RxJava 的事件回调方法除了普通事件 onNext()(相当于 onClick() / onEvent())之外,还定义了两个特殊的事件:onCompleted()和 onError()。
  onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理,还会把它们看做一个队列。RxJava 规定,当不会再有新的onNext()发出时,需要触发 onCompleted()方法作为标志。
  onError(): 事件队列异常。在事件处理过程中出异常时,onError()会被触发,同时队列自动终止,不允许再有事件发出。
在一个正确运行的事件序列中, onCompleted()和 onError()有且只有一个,并且是事件序列中的最后一个。需要注意的是,onCompleted()
和 onError()二者也是互斥的,即在队列中调用了其中一个,就不应该再调用另一个。

RxJava 的观察者模式大致如下图:


RxJava的观察者模式(二)_第3张图片
Paste_Image.png

三、基本实现

1、创建Observer

Observer 即观察者,它决定事件触发的时候将有怎样的行为。 RxJava 中的 Observer接口的实现方式:

Observer observer = new Observer() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

除了 Observer 接口之外,RxJava 还内置了一个实现了 Observer的抽象类:Subscriber。 Subscriber对 Observer接口进行了一些扩展,但他们的基本使用方式是完全一样的:

Subscriber subscriber = new Subscriber() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

不仅基本使用方式一样,实质上,在 RxJava 的 subscribe 过程中,Observer也总是会先被转换成一个 Subscriber再使用。所以如果你只想使用基本功能,选择 Observer和 Subscriber是完全一样的。它们的区别对于使用者来说主要有两点:
  onStart(): 这是 Subscriber增加的方法。它会在 subscribe 刚开始,而事件还未发送之前被调用,可以用于做一些准备工作,例如数据的清零或重置。这是一个可选方法,默认情况下它的实现为空。需要注意的是,如果对准备工作的线程有要求(例如弹出一个显示进度的对话框,这必须在主线程执行), onStart()就不适用了,因为它总是在 subscribe 所发生的线程被调用,而不能指定线程。要在指定的线程来做准备工作,可以使用 doOnSubscribe()方法,具体可以在后面的文中看到。
  unsubscribe(): 这是 Subscriber所实现的另一个接口 Subscription的方法,用于取消订阅。在这个方法被调用后,Subscriber将不再接收事件。一般在这个方法调用前,可以使用 isUnsubscribed()先判断一下状态。 unsubscribe()这个方法很重要,因为在 subscribe()之后, Observable会持有 Subscriber的引用,这个引用如果不能及时被释放,将有内存泄露的风险。所以最好保持一个原则:要在不再使用的时候尽快在合适的地方(例如 onPause()onStop()等方法中)调用unsubscribe()来解除引用关系,以避免内存泄露的发生。

2、 创建 Observable

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

Observable observable = Observable.create(new Observable.OnSubscribe() {
    @Override
    public void call(Subscriber subscriber) {
        subscriber.onNext("Hello");
        subscriber.onNext("Hi");
        subscriber.onNext("Aloha");
        subscriber.onCompleted();
    }
});

可以看到,这里传入了一个 OnSubscribe对象作为参数。OnSubscribe会被存储在返回的 Observable对象中,它的作用相当于一个计划表,当 Observable被订阅的时候,OnSubscribe的 call()方法会自动被调用,事件序列就会依照设定依次触发(对于上面的代码,就是观察者Subscriber将会被调用三次 onNext()和一次 onCompleted())。这样,由被观察者调用了观察者的回调方法,就实现了由被观察者向观察者的事件传递,即观察者模式。
  create()方法是 RxJava 最基本的创造事件序列的方法。基于这个方法, RxJava 还提供了一些方法用来快捷创建事件队列,例如:

just(T...): 将传入的参数依次发送出来。

Observable observable = Observable.just("Hello", "Hi", "Aloha");
// 将会依次调用:
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

from(T[])/ from(Iterable): 将传入的数组或 Iterable拆分成具体对象后,依次发送出来。

String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
// 将会依次调用:
// onNext("Hello");
// onNext("Hi");
// onNext("Aloha");
// onCompleted();

上面 just(T...)的例子和 from(T[])的例子,都和之前的 create(OnSubscribe)的例子是等价的。

3、Subscribe (订阅)

创建了 Observable和 Observer之后,再用 subscribe()方法将它们联结起来,整条链子就可以工作了。代码形式很简单:

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

除了 subscribe(Observer)和 subscribe(Subscriber),subscribe()还支持不完整定义的回调,RxJava 会自动根据定义创建出Subscriber 。形式如下:

Action1 onNextAction = new Action1() {
    // onNext()
    @Override
    public void call(String s) {
        Log.d(tag, s);
    }
};
Action1 onErrorAction = new Action1() {
    // 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);

简单解释一下这段代码中出现的 Action1和 Action0。 Action0是 RxJava 的一个接口,它只有一个方法 call(),这个方法是无参无返回值的;由于 onCompleted()方法也是无参无返回值的,因此 Action0可以被当成一个包装对象,将 onCompleted()的内容打包起来将自己作为一个参数传入 subscribe()以实现不完整定义的回调。这样其实也可以看做将 onCompleted() 方法作为参数传进了subscribe(),相当于其他某些语言中的『闭包』。
  Action1也是一个接口,它同样只有一个方法 call(T param),这个方法也无返回值,但有一个参数;与 Action0同理,由于 onNext(T obj)和 onError(Throwable error)也是单参数无返回值的,因此 Action1可以将 onNext(obj)和 onError(error)打包起来传入 subscribe()以实现不完整定义的回调。事实上,虽然 Action0和 Action1在 API 中使用最广泛,但 RxJava 是提供了多个 ActionX形式的接口 (例如 Action2, Action3) 的,它们可以被用以包装不同的无返回值的方法。

三、场景示例

1. 打印字符串数组

将字符串数组 names中的所有字符串依次打印出来:

String[] names = ...;
Observable.from(names)
    .subscribe(new Action1() {
        @Override
        public void call(String name) {
            Log.d(tag, name);
        }
    });

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

你可能感兴趣的