RxJava源码解析02-订阅过程

柔情只为你懂 2023-10-17 15:28 144阅读 0赞

### RxJava源码解析-订阅过程 ###

以最简单的调用过程为例

Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            subscriber.onNext("hello");
            subscriber.onNext("world");
            subscriber.onCompleted();
        }
    }).subscribe(new Subscriber<String>() {
        @Override
        public void onCompleted() {
    
        }
    
        @Override
        public void onError(Throwable e) {
    
        }
    
        @Override
        public void onNext(String s) {
    
        }  
    }

#### create() ####

先来看`create()`

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

此处RxJavaHooks.onCreate(f)返回的就是OnSubscribe对象，然后作为传参传入Observable并返回。

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

**即create会将f赋值给Observable.onSubscribe，并返回Observable对象**

#### subscribe() ####

再来看subscribe

public final Subscription subscribe(Subscriber<? super T> subscriber) {
        return Observable.subscribe(subscriber, this);
    }  
    
    static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
        subscriber.onStart();
        //...
        RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);
        return RxJavaHooks.onObservableReturn(subscriber);
    }

此处首先会调用`subscriber.onStart()`，  
接着`RxJavaHooks.onObservableStart`一般实际返回的就是observable.onSubscribe，  
然后调用`observable.onSubscribe.call()`，  
最后通过`RxJavaHooks.onObservableReturn`返回`Subscription`对象。(也就是subscriber，Subscriber实现了Subscription接口)

#### 总结 ####

Observable.subscribe()的过程其实就是先调用subscriber.onStart()，再通过Observable.OnSubscribe.call(subscriber)来间接调用subscriber的onNext、onError、onCompleted等方法。call的具体实现各个OnSubscribe是不同的。

Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            subscriber.onNext("hello world");
            subscriber.onNext("tttt");
            subscriber.onCompleted();
        }
    })