RxJava一次的create订阅执行流程

RxJava最简单的create使用流程如下，订阅前未使用任何变换：

Observable.create(new Observable.OnSubscribe<Integer>() {
     @Override
     public void call(Subscriber<? super Integer> subscriber) {
         subscriber.onNext(1);
         subscriber.onCompleted();
     }
 }).subscribe(new Subscriber<Integer>() {
     @Override
     public void onCompleted() {

     }

     @Override
     public void onError(Throwable e) {

     }

     @Override
     public void onNext(Integer integer) {

     }
 });

在分析源码前，我们用最简单的方式模拟一次create的执行流程。为了方便理解，使用的类名和RxJava中的类名保持一致，同时添加了一个Simple前缀作为区分：

定义SimpleSubscriber抽象类，为了简便，没有在onNext方法中添加参数：

public abstract class SimpleSubscriber {
    abstract void onCompleted();
    abstract void onError();
    abstract void onNext();
}

定义SimpleOnSubscribe接口，只有一个call方法，接受SimpleSubscriber作为参数：

public  interface SimpleOnSubscribe {
    void call(SimpleSubscriber subscriber);
}

然后我们定义SimpleObservable类，

public class SimpleObservable {

    private SimpleOnSubscribe simpleOnSubscribe;

    public SimpleObservable(SimpleOnSubscribe simpleOnSubscribe) {
        this.simpleOnSubscribe = simpleOnSubscribe;
    }

    public void subscribe(SimpleSubscriber subscriber){
        simpleOnSubscribe.call(subscriber);
    }

}

完成了三个类的定义，然后进行测试：

@Test
public void testCreate(){
    new SimpleObservable(new SimpleOnSubscribe() {
        @Override
        public void call(SimpleSubscriber subscriber) {
            subscriber.onNext();
            subscriber.onCompleted();
            subscriber.onError();
        }
    }).subscribe(new SimpleSubscriber() {
        @Override
        void onCompleted() {
            System.out.println("CreateTest.onCompleted");
        }

        @Override
        void onError() {
            System.out.println("CreateTest.onError");
        }

        @Override
        void onNext() {
            System.out.println("CreateTest.onNext");
        }
    });
}

输出：

CreateTest.onNext
CreateTest.onCompleted
CreateTest.onError

总结一下以上流程：

1. 构造Observable时，把OnSubscribe注入到
   Observable中
2. 调用Observable的subscirbe
3. Observable内部调用OnSubscribe的call方法
4. 把Subscriber作为参数传递给OnSubscribe的call方法。
5. 执行OnSubscribe的call方法，内部一次执行了

• CreateTest.onNext
• CreateTest.onCompleted
• CreateTest.onError

我们依照上述步骤来分析源码：

1. 构造Observable时，把OnSubscribe注入到
   Observable中

RxJavaHooks没有影响流程的执行，可以忽略其中的代码

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

final OnSubscribe<T> onSubscribe;

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

2. 调用Observable的subscirbe

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

3. 在Observable内部调用OnSubscribe的call方法
4. 把Subscriber作为参数传递给OnSubscribe的call方法。

static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
	//调用了call方法
	RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);
    return RxJavaHooks.onObservableReturn(subscriber);
	
}

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;
}

5. 执行OnSubscribe的call方法

这篇好文章是转载于：学新通技术网