前言

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

观察者模式

上一篇有提到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<Integer>，并在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