RxJava基本原理与使用(二)

581 阅读6分钟

###前言
  上一篇我们了解了RxJava的介绍以及通过一个例子学习了RxJava的展现形式,相信大家现在已经对RxJava有一个初步的印象与概念了。本篇我们将更深入的学习RxJava的基本原理以及如何使用RxJava。

###观察者模式
  上一篇有提到RxJava的核心便是扩展的观察者模式,我们先来学习一下观察者模式。

  事件:小明按下遥控器开关,空调启动并开始吹风了。按下制冷按键,空调吹出的风变为凉风。再按下3档风速按键,空调吹的风变大了。
  在这个事件中,空调作为观察者,遥控器作为被观察者,空调通过红外线检测遥控器传来的信号并作出响应。

  通过上图我们可以很清晰的看出遥控器作为事件的生产方主动发起事件,是事件的起点,而空调作为事件的处理方被动接收事件做出响应,是事件的终点。而在事件传递的过程中可对事件进行过滤,转换,合并等操作,就像例子中制冷与加大风速的效果。
  这个概念是RxJava中的核心,它是基于观察者模式来组建自己的程序逻辑,遥控器就相当于RxJava的被观察者Observable,空调相当于观察者Observer,而对事件处理的一些操作相当于各类操作符。不同的是RxJava中观察者需要对被观察者进行订阅,这样当被观察者发出事件时,观察者才会接收到事件进行处理。若未进行订阅,被观察者是不会发出任何事件的。

使用

  了解了观察者模式,接下来动起指尖尝试下怎么使用RxJava。

创建被观察者Observable

  第一步:创建被观察者Observable:

Observable<Object> observable = Observable.create(new Observable.OnSubscribe<Object>() {
    @Override
    public void call(Subscriber<? super Object> subscriber) {
        //被观察者的数据操作更新
        for (int i = 0; i < 5; i++)
            subscriber.onNext("xulei" + i);
        }
        subscriber.onCompleted();
    }
});

  通过Observable.create()方法创建并返回一个Observable对象,create()方法需要一个OnSubsribe对象作为参数传入并执行call()方法。此OnSubsribe对象继承自Action1。
  而在call()方法中可通过subscriber.onNext()进行事件的发送,subscriber.onCompleted()标记为事件发送完毕。subscriber.onError()标记为事件发送过程出现异常。
  除create()方法创建Observable方式外,还可以使用just()与from()等方式,这个在之后的创建操作符文章中会有讲解。

创建观察者Observer

  第二步:创建观察者Observer:

Observer observer = new Observer() {
    @Override
    public void onCompleted() { //标记事件发送完毕的回调
        Log.e("rx_test", "onCompleted");
    }
    @Override
    public void onError(Throwable e) { //事件发送过程出现异常的回调
        Log.e("rx_test", "onError");
    }
    @Override
    public void onNext(Object o) { //接收到发送事件的回调
        Log.e("rx_test", "onNext:" + o.toString());
    }
};

  在以往普通的观察者模式中,观察者通常只提供一个update()的方法,而RxJava扩展的观察者模式中的观察者提供onNext(Object o)onCompleted()onError(Throwable e)三个方法,分别对应被观察者call()方法中重写的subscriber.onNext()subscriber.onCompleted()subscriber.onError()方法。而onNext()方法就对应普通观察者模式中的update()方法。
  RxJava扩展的观察模式弥补了普通观察者模式的不足:
  1. 当事件发送完毕可以调用onCompleted()方法进行标示。
  2. 当事件发送过程中出现异常会自动触发onError()方法,也可进行手动调用。
  3. Observable支持链式编程,从而避免了回调嵌套的问题,也简化了代码。

订阅

  最后一步:被观察者Observable订阅观察者Observer。
  这里不同于普通观察者模式,上一篇我们讲了在RxJava扩展观察者模式中,如果没有观察者,被观察者是不会发出任何事件的。所以这里是被观察者订阅观察者。

observable.subscribe(observer);

  订阅很简单,一行代码搞定。调用Observable的subscribe()方法,传入观察者对象作为参数,这样被观察者与观察者就进行了绑定。

简写

  RxJava可以使用链式编程:

Observable.create(new Observable.OnSubscribe<Object>() {
    @Override
    public void call(Subscriber<? super Object> subscriber) {
        //被观察者的数据操作更新
        for (int i = 0; i < 5; i++) {
            subscriber.onNext("xulei" + i);
        }
        subscriber.onCompleted();
    }
}).subscribe(new Observer<Object>() {
    @Override
    public void onCompleted() {
        Log.e("rx_test", "onCompleted");
    }
    @Override
    public void onError(Throwable e) {
        Log.e("rx_test", "onError");
    }
    @Override
    public void onNext(Object o) {
        Log.e("rx_test", "onNext:" + o.toString());
    }
});

逻辑梳理

  到此一个RxJava的简单调用就实现了,相信不少码友看完之后还是很懵13,那我们再来梳理一下整个流程。首先使用Observable.create()创建了一个Observable,并在create()方法中传入了一个OnSubscribe对象作为参数,重写OnSubscribe中的call()方法,当Observable进行订阅后就会自动触发call()方法发送事件。
  注意:call()方法中的subscriber参数其实就是我们在调用subscribe()进行订阅时时传入的观察者Observer。所以当在call()方法中调用的5次onNext()方法与1次onCompleted()方法后,观察者Observer中输出的日志为:

onNext:xulei0
onNext:xulei1
onNext:xulei2
onNext:xulei3
onNext:xulei4
onCompleted

  有的码友看到这里会对call()方法中的subscriber参数就是订阅时传入的Observer有些不解,那么我们一起来看源码。
  先看下call()方法中的参数Subscriber:

public abstract class Subscriber<T> implements Observer<T>, Subscription {
    ......
}

  可以看出Subscriber是Observer的抽象实现类,与Observer一个类型。再来看订阅时调用的subscribe(observer)方法:

public final Subscription subscribe(final Observer<? super T> observer) {
    //无视
    if (observer instanceof Subscriber) {
        return subscribe((Subscriber<? super T>)observer);
    }
    return subscribe(new Subscriber<T>() {

        @Override
        public void onCompleted() {
            observer.onCompleted();
        }

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

        @Override
        public void onNext(T t) {
            observer.onNext(t);
        }
    });
}

  这里将传入的Observer对象通过代理转换为Subscriber对象作为参数传入并调用重载方法subscribe(),我们再来看这个subscribe()方法的代码:

public final Subscription subscribe(Subscriber<? super T> subscriber) {
    return Observable.subscribe(subscriber, this);
}

  这里继续调用重载方法subscribe(),将由Observer转换为的subscriber对象与当前Observable对象作为参数传入,继续往下追踪源码:

private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
    ......
    subscriber.onStart();
    ......
    try {
        hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
        return hook.onSubscribeReturn(subscriber);
    } catch (Throwable e) {
        ......
        return Subscriptions.unsubscribed();
    }
}

  精简一下代码发现,subscriber.onStart()用来通知被观察者准备开始发送事件了。重点看hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);我们发现hook.onSubscribeStart(observable, observable.onSubscribe)方法返回的是其第二个参数observable.onSubscribe,这个参数正是新建Observable时create()方法中传入的OnSubscribe对象。接着又通过这个onSubscribe调用call(subscriber)方法,将subscriber作为参数传入,码友们有没有发现,这个subscriber传来传去究其根源,正是一开始由订阅方法subscribe(observer)传入的observer通过代理转换为的那个subscriber。
  这样前后逻辑就连接上了,通过阅读源码我们才能发现为何创建Observable时call()方法中的subscriber参数其实就是订阅时传入的Observer。

总结

  到此,本篇关于RxJava的基本原理与使用就讲解完毕了,下一篇我们将一起研究RxJava的四类操作符中的创建操作符都有哪些以及如何使用。
  这里感谢下张磊大神撰写的RxJava系列的帮助:
  RxJava系列2(基本概念及使用介绍)
  技术渣一枚,有写的不对的地方欢迎大神们留言指正,有什么疑惑或者建议也可以在我Github上RxJavaDemo项目Issues中提出,我会及时回复。
  附上RxJavaDemo的地址:
  RxJavaDemo