Rxjava源码解析笔记 | 创建Observable 与 Observer/Subscriber 以及之间订阅实现的源码分析

简单回顾

如果抛开Rxjava的操作符以及其线程控制的话，Rxjava的最基本使用是比较简单的
第一步，创建被观察者Observable；
第二步，创建观察者Observer/Subscriber；
第三步，subscribe；

三个关键对象和一个核心方法

• 三个关键对象

  • Observable（被观察者）
  • OnSubscribe

此对象是Observable内部的对象，
可以对等地理解成是观察者模式中，
被观察者用来通知观察者的notifyObservers()方法；

- Subscriber（观察者）

• 一个核心方法

  • subscribe()

订阅方法，
完成观察者和被观察者之间的订阅；

Rxjava源码相关分析

Observable类中的create()源码：

public class Observable<T> {

    final OnSubscribe<T> onSubscribe;

    protected Observable(OnSubscribe<T> f) {
        this.onSubscribe = f;
    }

    static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();

    public static <T> Observable<T> create(OnSubscribe<T> f) {
        return new Observable<T>(hook.onCreate(f));
    }

    @Beta
    public static <S, T> Observable<T> create(SyncOnSubscribe<S, T> syncOnSubscribe) {
        return new Observable<T>(hook.onCreate(syncOnSubscribe));
    }

    @Experimental
    public static <S, T> Observable<T> create(AsyncOnSubscribe<S, T> asyncOnSubscribe) {
        return new Observable<T>(hook.onCreate(asyncOnSubscribe));
    }

    ...

}

• 第一步，调用Observable.create()方法完成Observable的创建；

下面跟进create()源码，

- 我们发现 ```create方法内部```，

会传入一个OnSubscribe的参数，
其实这个从外部传进来的OnSubscribe参数，
最终也会通过new调用Observable的构造方法，
最终传给Observable的全局变量 onSubscribe

- 而关于```全局变量 Observable.OnSubscribe类型的对象 onSubscribe```，

我们刚刚提到，
可以对等地理解成是观察者模式中，
被观察者用来通知观察者的notifyObservers()方法；

- 接着我们看源码中第一个```create()```的重载方法，

其最后都是return，返回一个new调用构造方法创建好的Observable对象；

- 而在```create()```内部，

赋值给构造函数的参数是调用了一个hook.onCreate()方法；

下面关注一下hook，
可以看到关于hook的定义——
static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();

事实上我们可以把hook理解成为一个抽象的代理类对象；
默认情况下，hook不会的OnSubscribe对象做任何的处理；

**一句话总结一下，
Observable的就是通过代理类对象hook创建的，
而默认情况下，hook不会的OnSubscribe对象做任何的处理；**
当然，默认不处理，
但是我们需要的话自然是可以个性化地重写这个方法的；

**所以create()所做的工作是，
构造一个新的Observable对象，
同时借助Observable类的原生构造方法，
将传进来的OnSubscribe对象赋给类中的全局变量onSubscribe；
如此便完成了被观察者Observable的创建生成；**

示例代码如下：

//第一步：创建被观察者：create
   Observable observable = Observable.create(new >Observable.OnSubscribe<String>() {
       @Override
       public void call(Subscriber<? super String> subscriber) {
           subscriber.onNext("Hello");
           subscriber.onNext("Imooc");
           subscriber.onCompleted();
       }
   });

   //通过just方法来创建被观察者
   Observable observableJust = Observable.just("hello", "Imooc");

   //通过from方法来创建被观察者
   String[] parameters = {"hello", "Imooc"};
   Observable observableFrom = Observable.from(parameters);

**可以看到，创建Observable的实质，
就是new一个OnSubscribe接口并实现其call()方法，
然后把实现了call()方法的整个OnSubscribe接口对象作为参数赋值给Observable的create()方法；

创建Observable 实例之后，
会把这个OnSubscribe 对象参数存到本Observable示例的全局变量中；

等到回调call()方法时候，
其实就是Observable.subscribe() - subscribe()中call()，**
**这个call()方法现在是“存储在”Observable 实例中的方法，
也是方才传进来的，
实现了call()方法的整个OnSubscribe接口对象参数；**

！！！这个地方实际上就跟onClick()的实现逻辑相当类似！！！！

观察者创建以及订阅实例代码：

    //第二步：创建观察者
    Observer<Object> observer = new Observer<Object>() {

        @Override
        public void onCompleted() {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onNext(Object s) {

        }
    };

    Subscriber<String> subscriber = new Subscriber<String>() {
        @Override
        public void onCompleted() {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onNext(String s) {

        }

        @Override
        public void onStart() {
            super.onStart();
        }
    };

    public void doRxjava(){
        //第三步：订阅
        observable.subscribe(observer);
        observable.subscribe(subscriber);
    }

• 第二步，创建Observer观察者对象；

这里有两个接口——Observer和Subscriber

- 其实```Subscriber```内部也只对```Observer```做了一些拓展，

它们两个的使用方式基本是相同的；
当然就是有一些回调接口的差异；

- Rxjava内部最终会将```Observer```转换成```Subscriber```；

接下来是Rxjava的SDK中subscribe()的传入参数
是Observer时候（observable.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);
            }

        });
    }

Rxjava的SDK中subscribe()的传入参数
是Subscriber时候（observable.subscribe(subscriber);）的源码：

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

• .

  • 这里可以看到如果传给subscribe()的对象是Observer，

则会在源码Rxjava的源码中首先被转换成Subscriber，
之后再进行后续操作；
**也即应证了之前所说的Rxjava内部

最终会将Observer转换成Subscriber再进行后续处理；**

Subscriber源码分析

package rx;

import rx.internal.util.SubscriptionList;


public abstract class Subscriber<T> implements Observer<T>, Subscription {
    
    // represents requested not set yet
    private static final Long NOT_SET = Long.MIN_VALUE;

    private final SubscriptionList subscriptions;
    private final Subscriber<?> subscriber;
    /* protected by `this` */
    private Producer producer;
    /* protected by `this` */
    private long requested = NOT_SET; // default to not set

    protected Subscriber() {
        this(null, false);
    }

    protected Subscriber(Subscriber<?> subscriber) {
        this(subscriber, true);
    }

    protected Subscriber(Subscriber<?> subscriber, boolean shareSubscriptions) {
        this.subscriber = subscriber;
        this.subscriptions = shareSubscriptions && subscriber != null ? subscriber.subscriptions : new SubscriptionList();
    }

    public final void add(Subscription s) {
        subscriptions.add(s);
    }

    @Override
    public final void unsubscribe() {
        subscriptions.unsubscribe();
    }

    @Override
    public final boolean isUnsubscribed() {
        return subscriptions.isUnsubscribed();
    }

    public void onStart() {
        // do nothing by default
    }
    
    protected final void request(long n) {
        if (n < 0) {
            throw new IllegalArgumentException("number requested cannot be negative: " + n);
        } 
        
        // if producer is set then we will request from it
        // otherwise we increase the requested count by n
        Producer producerToRequestFrom = null;
        synchronized (this) {
            if (producer != null) {
                producerToRequestFrom = producer;
            } else {
                addToRequested(n);
                return;
            }
        }
        // after releasing lock (we should not make requests holding a lock)
        producerToRequestFrom.request(n);
    }

    private void addToRequested(long n) {
        if (requested == NOT_SET) {
            requested = n;
        } else { 
            final long total = requested + n;
            // check if overflow occurred
            if (total < 0) {
                requested = Long.MAX_VALUE;
            } else {
                requested = total;
            }
        }
    }

    public void setProducer(Producer p) {
        long toRequest;
        boolean passToSubscriber = false;
        synchronized (this) {
            toRequest = requested;
            producer = p;
            if (subscriber != null) {
                // middle operator ... we pass through unless a request has been made
                if (toRequest == NOT_SET) {
                    // we pass through to the next producer as nothing has been requested
                    passToSubscriber = true;
                }
            }
        }
        // do after releasing lock
        if (passToSubscriber) {
            subscriber.setProducer(producer);
        } else {
            // we execute the request with whatever has been requested (or Long.MAX_VALUE)
            if (toRequest == NOT_SET) {
                producer.request(Long.MAX_VALUE);
            } else {
                producer.request(toRequest);
            }
        }
    }
}

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

• Subscriber实现了Observer、Subscription两个接口；

Observer接口即前面所提的另一个观察者接口；

• Subscription源码

  • unsubscribe()：**解绑；

当“订阅事件的列表”
（也即当前观察者中的一个放着所有订阅事件的列表的成员变量）
之中不再有订阅事件时，
调用这个方法来对“订阅事件列表”进行解绑；**

- isUnsubscribed()：判断是否已经解绑订阅事件；

package rx;

import rx.subscriptions.Subscriptions;

public interface Subscription {
    void unsubscribe();

    boolean isUnsubscribed();
}

• private final SubscriptionList subscriptions;

这个List即前面unsubscribe()中提到的“订阅事件的列表”；
其中包含的是观察者所有的订阅事件；

当Subscriber 取消订阅的时候，
这个List中就会有事件被删除（得益于实现了Subscription接口）；

当这个List中没有任何事件了，
即列表中所有的事件都被取消订阅了；
那么这个List也就为空；

以上则是Subscriber的核心逻辑；

第三步，下面具体分析订阅的实现

**下面小结一下，call()方法在subscribe()方法之中被调用
（从下面subscribe()方法的源码中可以见得），

也即默认subscribe()方法被调用之后
（observable.subscribe(observer);或者observable.subscribe(subscriber);），

会触发在创建Observable时候实现的OnSubscribe<>中的call()方法，
完成回调；
call()方法中可以实现一系列事件消费的过程——onNext()、onCompleted()等；**
observable中调用了onNext、onCompleted()的时候，
相应订阅了的Observer/Subscriber中的onNext、onCompleted()也就被调用执行；

//第一步：创建被观察者：create
   Observable observable = Observable.create(new >Observable.OnSubscribe<String>() {
       @Override
       public void call(Subscriber<? super String> subscriber) {
           subscriber.onNext("Hello");
           subscriber.onNext("Imooc");
           subscriber.onCompleted();
       }
   });

   //通过just方法来创建被观察者
   Observable observableJust = Observable.just("hello", "Imooc");

   //通过from方法来创建被观察者
   String[] parameters = {"hello", "Imooc"};
   Observable observableFrom = Observable.from(parameters);

以上便是Rxjava最基本的使用方式；

Rxjava的SDK中subscribe()的传入参数
是Subscriber时候（observable.subscribe(subscriber);）的源码：

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

• 可以看到Subscription subscribe()中，

又调用了另外一个“同名的订阅方法”Observable.subscribe()，
传入的是subscriber和this两个参数；

• 跟进Observable.subscribe()的源码：

    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.");
        }
        
        subscriber.onStart();

        if (!(subscriber instanceof SafeSubscriber)) {
            subscriber = new SafeSubscriber<T>(subscriber);
        }

        try {
            hook.onSubscribeStart(observable, observable.onSubscribe).call(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();
        }
    }

• 首先是对subscriber和observable.onSubscribe判空，

空则抛出异常；
if (subscriber == null) {... }
if (observable.onSubscribe == null) {...}

接着，

• subscriber.onStart();，

跟进一下onStart()，
可以发现它其实是Subscriber类中的一个空方法；
默认什么都不做，需要我们自己调用的时候去实现；

• 要把传进来的subscriber包装成SafeSubscriber来进行后续操作，

而这个SafeSubscriber其实就是比subscriber多做了一系列更完善的处理，
当然它也包含了subscriber拥有的onCompleted()、onError()等方法：

if (!(subscriber instanceof SafeSubscriber)) {
            subscriber = new SafeSubscriber<T>(subscriber);
        }

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

return hook.onSubscribeReturn(subscriber);
接着我们可以看到源码以subscriber为参数，
调用了call()回调方法；
**调用完了call()方法，
意味着完成了一次订阅；**
接着通过代理类hook将subscriber以Subscription的类型返回
（public <T> Subscription onSubscribeReturn(Subscription subscription) {...}）
（此处subscriber对象向上转型，
当成接口Subscription类型使用，
因为Subscriber类实现了接口Subscription，乃其子类）；

前面提到过，
接口Subscription是每一个观察者Subscriber必须实现的接口，
同时Subscriber的实例都有一个SubscriptionList，
这个List包含的是观察者所有的订阅事件；

至此便完成了Subscriber核心的主要的订阅流程；

• 跟进SafeSubscriber源码：

public class SafeSubscriber<T> extends Subscriber<T> {

    private final Subscriber<? super T> actual;

    boolean done = false;

    public SafeSubscriber(Subscriber<? super T> actual) {
        super(actual);
        this.actual = actual;
    }

    @Override
    public void onCompleted() {
        if (!done) {
            done = true;
            try {
                actual.onCompleted();
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                RxJavaPluginUtils.handleException(e);
                throw new OnCompletedFailedException(e.getMessage(), e);
            } finally {
                try {
                    unsubscribe();
                } catch (Throwable e) {
                    RxJavaPluginUtils.handleException(e);
                    throw new UnsubscribeFailedException(e.getMessage(), e);
                }
            }
        }
    }

    @Override
    public void onError(Throwable e) {
        Exceptions.throwIfFatal(e);
        if (!done) {
            done = true;
            _onError(e);
        }
    }


    @Override
    public void onNext(T args) {
        try {
            if (!done) {
                actual.onNext(args);
            }
        } catch (Throwable e) {
            Exceptions.throwOrReport(e, this);
        }
    }

    protected void _onError(Throwable e) {
        RxJavaPluginUtils.handleException(e);
        try {
            actual.onError(e);
        } catch (Throwable e2) {
            if (e2 instanceof OnErrorNotImplementedException) {
                try {
                    unsubscribe();
                } catch (Throwable unsubscribeException) {
                    RxJavaPluginUtils.handleException(unsubscribeException);
                    throw new RuntimeException("Observer.onError not implemented and error while unsubscribing.", new CompositeException(Arrays.asList(e, unsubscribeException)));
                }
                throw (OnErrorNotImplementedException) e2;
            } else {
                RxJavaPluginUtils.handleException(e2);
                try {
                    unsubscribe();
                } catch (Throwable unsubscribeException) {
                    RxJavaPluginUtils.handleException(unsubscribeException);
                    throw new OnErrorFailedException("Error occurred when trying to propagate error to Observer.onError and during unsubscription.", new CompositeException(Arrays.asList(e, e2, unsubscribeException)));
                }

                throw new OnErrorFailedException("Error occurred when trying to propagate error to Observer.onError", new CompositeException(Arrays.asList(e, e2)));
            }
        }
        try {
            unsubscribe();
        } catch (RuntimeException unsubscribeException) {
            RxJavaPluginUtils.handleException(unsubscribeException);
            throw new OnErrorFailedException(unsubscribeException);
        }
    }

    public Subscriber<? super T> getActual() {
        return actual;
    }
}

• 从源码可以看到SafeSubscriber类中，

也有onCompleted() 、onError()、onNext()等方法；
它就是对subscriber类做了一个包装；

• 注意onCompleted() 或者onError()方法被执行的时候，

就不会再执行onNext()（注意布尔变量done）；
（因完成状态/Completed或者异常状态/Error发生后，
就没必要在进行下一步/Next的操作了）
（onComplete()和onError()是互斥的，
也就是一次只能调用其中一个，不能同时调用）

参考自 慕课网