RxJava源码解析（1） —— subscribe

RxJava是一款实现异步的库，因为具备比较高的抽象性，使用它能够让咱们的项目代码逻辑更加的清晰，从而更利于维护。

今天咱们分析下RxJava的最基础的部分：subscribe()的实现。

首先看RxJava最基本的使用方式：

Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("Hello world");
                subscriber.onCompleted();
            }
        }).subscribe(new Subscriber<String>() {
            @Override
            public void onStart() {
                super.onStart();
            }
            @Override
            public void onCompleted() {
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onNext(String o) {
                System.out.println(o);
            }
        });

上述代码很简单，就是经过Observable.create新建了一个Observable 而后 订阅（subscribe） 一个Subscriber。

下面咱们就来看看subscribe的内部究竟做了哪些事？

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

 

private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
  
        //合法性校验
        if (subscriber == null) {
            throw new IllegalArgumentException("observer can not be null");
        }
        if (observable.onSubscribe == null) {
            throw new IllegalStateException("onSubscribe function can not be null.");
        }
        //onStart()回调
        subscriber.onStart();

        //包装成SafeSubscriber
        if (!(subscriber instanceof SafeSubscriber)) {
            subscriber = new SafeSubscriber<T>(subscriber);
        }

        //调用observable.onSubscribe.call()方法
        try {
            hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
         //返回subscriber
            return hook.onSubscribeReturn(subscriber);
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            try {
                subscriber.onError(hook.onSubscribeError(e));
            } catch (Throwable e2) {
                Exceptions.throwIfFatal(e2);
                RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
                hook.onSubscribeError(r);
                throw r;
            }
            //解绑
            return Subscriptions.unsubscribed();
        }
    }

由上可知最关键的代码在这一句：

 hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);

而观察源码能够知道：

 public <T> OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance, final OnSubscribe<T> onSubscribe) {
        // pass-thru by default
        return onSubscribe;
    }

因此上述代码等价于：

observable.onSubscribe.call(subscriber);实际上就是把subscriber做为onSubscribe.call（）方法的参数传了进去，而这里的onSubscribe就是Observale.create的时候new出来的那个，因此在subscribe的过程当中会依次调用subscriber.onNext  和 onComplete方法。至此，就很明确了，订阅subscribe的过程就是把最终的subscriber做为Observable内部onSubscribe的call方法参数传入，最终在onSubscribe的call方法内部不断的调用subscriber的回调方法。这是典型的观察者模式。