Rxjava源码解析笔记 | 创建Observable 与 Ob

2019-06-04 本文已影响8人凌川江雪

简单回顾

如果抛开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的创建生成；

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

    //第二步：创建观察者
    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()等；
//第一步：创建被观察者：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的操作了）

参考自慕课网

Rxjava源码解析笔记 | 创建Observable 与 Ob

简单回顾

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

Rxjava源码相关分析

`Observable`类中的`create()源码`：

Subscriber源码分析

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

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

跟进`SafeSubscriber`源码：

猜你喜欢

热点阅读

Rxjava源码解析笔记 | 创建Observable 与 Ob

简单回顾

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

Rxjava源码相关分析

Observable类中的create()源码：

Subscriber源码分析

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

跟进Observable.subscribe()的源码：

跟进SafeSubscriber源码：

猜你喜欢

热点阅读

`Observable`类中的`create()源码`：

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

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

跟进`SafeSubscriber`源码：