当前位置：首页 > news >正文

RxJava的订阅过程

news 2025/8/24 15:05:50

要使用Rxjava首先要导入两个包，其中rxandroid是rxjava在android中的扩展

 	 implementation 'io.reactivex:rxandroid:1.2.1'implementation 'io.reactivex:rxjava:1.2.0'

首先从最基本的Observable的创建到订阅开始分析

    Observable.create(new Observable.OnSubscribe<String>() {@Overridepublic void call(Subscriber<? super String> subscriber) {subscriber.onNext("observable call onNext0");subscriber.onStart();subscriber.onNext("observable call onNext");subscriber.onCompleted();subscriber.onNext("observable call onNext1");}}).subscribe(new Subscriber<String>() {@Overridepublic void onCompleted() {ILog.LogDebug("subscriber onCompleted");}@Overridepublic void onError(Throwable e) {ILog.LogDebug("subscriber onError");}@Overridepublic void onNext(String s) {ILog.LogDebug("subscriber onNext");}});

Observable.create()需要一个OnSubscribe，OnSubscribe又是什么呢

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

OnSubscribe是一个接口，继承自Action1，Action1继承自Action，Action继承自Function，Function就是所有的action和fun的基类，于是有
OnSubscribe > Action1 > Action > Function , 由于Action1 接口有一个call方法，OnSubscribe接口也拥有了一个call方法。call方法的参数是一个Subscriber

    /*** Invoked when Observable.subscribe is called.* @param <T> the output value type*/public interface OnSubscribe<T> extends Action1<Subscriber<? super T>> {// cover for generics insanity}/*** A one-argument action.* @param <T> the first argument type*/public interface Action1<T> extends Action {void call(T t);}/*** All Action interfaces extend from this.* <p>* Marker interface to allow instanceof checks.*/public interface Action extends Function {}/*** All Func and Action interfaces extend from this.* <p>* Marker interface to allow instanceof checks.*/public interface Function {}

接着继续看RxJavaHooks.onCreate(f)做了什么 , 由于RxJavaHooks 源码较多，这里只贴了关键的一部分，onObservableCreate为RxJavaHooks 初始化后在static区自动执行赋值的，Func1类型，RxJavaHooks.onCreate最后也就是调用了f.call(onSubscribe)，参数是我们传进去的onSubscribe实例

public final class RxJavaHooks {........@SuppressWarnings("rawtypes")static volatile Func1<Observable.OnSubscribe, Observable.OnSubscribe> onObservableCreate;@SuppressWarnings("rawtypes")static volatile Func2<Observable, Observable.OnSubscribe, Observable.OnSubscribe> onObservableStart;static volatile Func1<Subscription, Subscription> onObservableReturn;
......../*** Hook to call when an Observable is created.* @param <T> the value type* @param onSubscribe the original OnSubscribe logic* @return the original or replacement OnSubscribe instance*/@SuppressWarnings({ "rawtypes", "unchecked" })public static <T> Observable.OnSubscribe<T> onCreate(Observable.OnSubscribe<T> onSubscribe) {Func1<Observable.OnSubscribe, Observable.OnSubscribe> f = onObservableCreate;if (f != null) {return f.call(onSubscribe);}return onSubscribe;}/*** Hook to call before the Observable.subscribe() method is about to return a Subscription.* @param subscription the original subscription* @return the original or alternative subscription that will be returned*/public static Subscription onObservableReturn(Subscription subscription) {Func1<Subscription, Subscription> f = onObservableReturn;if (f != null) {return f.call(subscription);}return subscription;}/*** Hook to call before the child subscriber is subscribed to the OnSubscribe action.* @param <T> the value type* @param instance the parent Observable instance* @param onSubscribe the original OnSubscribe action* @return the original or alternative action that will be subscribed to*/@SuppressWarnings({ "rawtypes", "unchecked" })public static <T> Observable.OnSubscribe<T> onObservableStart(Observable<T> instance, Observable.OnSubscribe<T> onSubscribe) {Func2<Observable, Observable.OnSubscribe, Observable.OnSubscribe> f = onObservableStart;if (f != null) {return f.call(instance, onSubscribe);}return onSubscribe;}.......@SuppressWarnings({ "rawtypes", "unchecked", "deprecation" })static void initCreate() {onObservableCreate = new Func1<Observable.OnSubscribe, Observable.OnSubscribe>() {@Overridepublic Observable.OnSubscribe call(Observable.OnSubscribe f) {return RxJavaPlugins.getInstance().getObservableExecutionHook().onCreate(f);}};onObservableStart = new Func2<Observable, Observable.OnSubscribe, Observable.OnSubscribe>() {@Overridepublic Observable.OnSubscribe call(Observable t1, Observable.OnSubscribe t2) {return RxJavaPlugins.getInstance().getObservableExecutionHook().onSubscribeStart(t1, t2);}};onObservableReturn = new Func1<Subscription, Subscription>() {@Overridepublic Subscription call(Subscription f) {return RxJavaPlugins.getInstance().getObservableExecutionHook().onSubscribeReturn(f);}};onSingleCreate = new Func1<rx.Single.OnSubscribe, rx.Single.OnSubscribe>() {@Overridepublic rx.Single.OnSubscribe call(rx.Single.OnSubscribe f) {return RxJavaPlugins.getInstance().getSingleExecutionHook().onCreate(f);}};onCompletableCreate = new Func1<Completable.OnSubscribe, Completable.OnSubscribe>() {@Overridepublic Completable.OnSubscribe call(Completable.OnSubscribe f) {return RxJavaPlugins.getInstance().getCompletableExecutionHook().onCreate(f);}};}....}

在 f.call中又调用了
RxJavaPlugins.getInstance().getObservableExecutionHook().onCreate(f); 最后返回的就是我们传入的onSubscribe

public abstract class RxJavaObservableExecutionHook { ......@Deprecatedpublic <T> OnSubscribe<T> onCreate(OnSubscribe<T> f) {return f;}@Deprecatedpublic <T> OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance, final OnSubscribe<T> onSubscribe) {// pass through by defaultreturn onSubscribe;}@Deprecatedpublic <T> Subscription onSubscribeReturn(Subscription subscription) {// pass through by defaultreturn subscription;}
......
}

最后在回来看new Observable(RxJavaHooks.onCreate(f))， Observable 的构造方法，Observable把传入的onSubscribe 保存了起来。至此饶了一大圈Observable对象产生。

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

下面会继续调用Observable的subscribe方法并传入Observer（观察者），完成订阅操作。现在来查看Observable的subscribe方法做了什么

public class Observable<T> {
.....public final Subscription subscribe(Subscriber<? super T> subscriber) {return Observable.subscribe(subscriber, this);}
.....static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {// validate and proceedif (subscriber == null) {throw new IllegalArgumentException("subscriber can not be null");}if (observable.onSubscribe == null) {  //此处的onSubscribe正是我们创建订阅的时候传入的onSubscribe throw new IllegalStateException("onSubscribe function can not be null.");/** the subscribe function can also be overridden but generally that's not the appropriate approach* so I won't mention that in the exception*/}// new Subscriber so onStart itsubscriber.onStart();/** See https://github.com/ReactiveX/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls* to user code from within an Observer"*/// if not already wrappedif (!(subscriber instanceof SafeSubscriber)) {// assign to `observer` so we return the protected versionsubscriber = new SafeSubscriber<T>(subscriber);}// The code below is exactly the same an unsafeSubscribe but not used because it would// add a significant depth to already huge call stacks.try {// allow the hook to intercept and/or decorateRxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);return RxJavaHooks.onObservableReturn(subscriber);} catch (Throwable e) {// special handling for certain Throwable/Error/Exception typesExceptions.throwIfFatal(e);// in case the subscriber can't listen to exceptions anymoreif (subscriber.isUnsubscribed()) {RxJavaHooks.onError(RxJavaHooks.onObservableError(e));} else {// if an unhandled error occurs executing the onSubscribe we will propagate ittry {subscriber.onError(RxJavaHooks.onObservableError(e));} catch (Throwable e2) {Exceptions.throwIfFatal(e2);// if this happens it means the onError itself failed (perhaps an invalid function implementation)// so we are unable to propagate the error correctly and will just throwRuntimeException r = new OnErrorFailedException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);// TODO could the hook be the cause of the error in the on error handling.RxJavaHooks.onObservableError(r);// TODO why aren't we throwing the hook's return value.throw r; // NOPMD}}return Subscriptions.unsubscribed();}}.....
}

在正式的订阅关系产生之前，首先会执行subscriber.onStart()方法，这里可以做一些初始化工作。继续往下看又判断是subscriber 实例是否是一个SafeSubscriber，不是则会新建一个SafeSubscriber来包装subscriber

       if (!(subscriber instanceof SafeSubscriber)) {// assign to `observer` so we return the protected versionsubscriber = new SafeSubscriber<T>(subscriber);}

老规矩，查看SafeSubscriber源码

public class SafeSubscriber<T> extends Subscriber<T> {private final Subscriber<? super T> actual;boolean done;public SafeSubscriber(Subscriber<? super T> actual) {super(actual);this.actual = actual;}/*** Notifies the Subscriber that the {@code Observable} has finished sending push-based notifications.* <p>* The {@code Observable} will not call this method if it calls {@link #onError}.*/@Overridepublic void onCompleted() {if (!done) {done = true;try {actual.onCompleted();} catch (Throwable e) {// we handle here instead of another method so we don't add stacks to the frame// which can prevent it from being able to handle StackOverflowExceptions.throwIfFatal(e);RxJavaHooks.onError(e);throw new OnCompletedFailedException(e.getMessage(), e);} finally { // NOPMDtry {// Similarly to onError if failure occurs in unsubscribe then Rx contract is broken// and we throw an UnsubscribeFailureException.unsubscribe();} catch (Throwable e) {RxJavaHooks.onError(e);throw new UnsubscribeFailedException(e.getMessage(), e);}}}}/*** Notifies the Subscriber that the {@code Observable} has experienced an error condition.* <p>* If the {@code Observable} calls this method, it will not thereafter call {@link #onNext} or* {@link #onCompleted}.** @param e*          the exception encountered by the Observable*/@Overridepublic void onError(Throwable e) {// we handle here instead of another method so we don't add stacks to the frame// which can prevent it from being able to handle StackOverflowExceptions.throwIfFatal(e);if (!done) {done = true;_onError(e);}}/*** Provides the Subscriber with a new item to observe.* <p>* The {@code Observable} may call this method 0 or more times.* <p>* The {@code Observable} will not call this method again after it calls either {@link #onCompleted} or* {@link #onError}.** @param args*          the item emitted by the Observable*/@Overridepublic void onNext(T args) {try {if (!done) {actual.onNext(args);}} catch (Throwable e) {// we handle here instead of another method so we don't add stacks to the frame// which can prevent it from being able to handle StackOverflowExceptions.throwOrReport(e, this);}}/*** The logic for {@code onError} without the {@code isFinished} check so it can be called from within* {@code onCompleted}.** @see <a href="https://github.com/ReactiveX/RxJava/issues/630">the report of this bug</a>*/protected void _onError(Throwable e) { // NOPMDRxJavaHooks.onError(e);try {actual.onError(e);} catch (OnErrorNotImplementedException e2) { // NOPMD/** onError isn't implemented so throw** https://github.com/ReactiveX/RxJava/issues/198** Rx Design Guidelines 5.2** "when calling the Subscribe method that only has an onNext argument, the OnError behavior* will be to rethrow the exception on the thread that the message comes out from the observable* sequence. The OnCompleted behavior in this case is to do nothing."*/try {unsubscribe();} catch (Throwable unsubscribeException) {RxJavaHooks.onError(unsubscribeException);throw new OnErrorNotImplementedException("Observer.onError not implemented and error while unsubscribing.", new CompositeException(Arrays.asList(e, unsubscribeException))); // NOPMD}throw e2;} catch (Throwable e2) {/** throw since the Rx contract is broken if onError failed** https://github.com/ReactiveX/RxJava/issues/198*/RxJavaHooks.onError(e2);try {unsubscribe();} catch (Throwable unsubscribeException) {RxJavaHooks.onError(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)));}// if we did not throw above we will unsubscribe here, if onError failed then unsubscribe happens in the catchtry {unsubscribe();} catch (Throwable unsubscribeException) {RxJavaHooks.onError(unsubscribeException);throw new OnErrorFailedException(unsubscribeException);}}/*** Returns the {@link Subscriber} underlying this {@code SafeSubscriber}.** @return the {@link Subscriber} that was used to create this {@code SafeSubscriber}*/public Subscriber<? super T> getActual() {return actual;}
}

SafeSubscriber是Subscriber的一个具体实现类，看SafeSubscriber像不像一个代理模式，具体的工作都是由actual来做，SafeSubscriber负责更完善的处理操作。
继续回到订阅部分的代码，类似之前的分析，代码已经在上面类贴出 RxJavaHooks.onObservableStart(observable, observable.onSubscribe)也只是返回了observable.onSubscribe实例，最后的.call(subscriber)也就是直接调用了我们在创建observable时传入的匿名实例call方法，最后返回subscriber。
RxJava的订阅过程就基本分析完了。

查看全文

http://www.lryc.cn/news/16703.html