代码示例

Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        LogUtils.loge("subscriber call ...");
        if (!subscriber.isUnsubscribed()) {
            subscriber.onNext("test1");
            subscriber.onCompleted();
        }
    }
});

Observer<String> observer = new Observer<String>() {

    @Override
    public void onCompleted() {
        LogUtils.loge("Observer onCompleted");
    }

    @Override
    public void onError(Throwable e) {
    }

    @Override
    public void onNext(String s) {
        LogUtils.loge("Observer onNext s = " + s);
    }
};
Subscription subscription = observable.subscribe(observer);

常用类说明

被观察者

rx.Observable

订阅

rx.Subscription

public interface Subscription {
    void unsubscribe();
    boolean isUnsubscribed();
}

观察者

rx.Observer

public interface Observer<T> {
    void onCompleted();
    void onError(Throwable e);
    void onNext(T t);
}

执行流程

rx.Observable#create(rx.Observable.OnSubscribe<T>)

public static <T> Observable<T> create(OnSubscribe<T> f) {          
    // 加载RxJavaHooks的static,初始化资源
    // 返回一个Observable对象
    return new Observable<T>(RxJavaHooks.onCreate(f));
}

rx.plugins.RxJavaHooks#onCreate(rx.Observable.OnSubscribe<T>)

public static <T> Observable.OnSubscribe<T> onCreate(Observable.OnSubscribe<T> onSubscribe) {
    Func1<Observable.OnSubscribe, Observable.OnSubscribe> f = onObservableCreate;
    if (f != null) {
        // 这里其实是调用到了onObservableCreate的call方法
        return f.call(onSubscribe);
    }
    return onSubscribe;
}

rx.plugins.RxJavaHooks

static {
    init();
}

static void init() {
    onObservableStart = new Func2<Observable, Observable.OnSubscribe, Observable.OnSubscribe>() {
        @Override
        public Observable.OnSubscribe call(Observable t1, Observable.OnSubscribe t2) {
            // 调用开始订阅的方法
            return RxJavaPlugins.getInstance().getObservableExecutionHook().onSubscribeStart(t1, t2);
        }
    };
    
    onObservableReturn = new Func1<Subscription, Subscription>() {
        @Override
        public Subscription call(Subscription f) {
            return RxJavaPlugins.getInstance().getObservableExecutionHook().onSubscribeReturn(f);
        }
    };
    
    initCreate();
}

static void initCreate() {
    onObservableCreate = new Func1<Observable.OnSubscribe, Observable.OnSubscribe>() {
        @Override
        public Observable.OnSubscribe call(Observable.OnSubscribe f) {
            /*
            这里1. 初始化RxJavaObservableExecutionHook 
                2. 返回我们传入的Observable.OnSubscribe
            */
            return RxJavaPlugins.getInstance().getObservableExecutionHook().onCreate(f);
        }
    };
}

rx.plugins.RxJavaPlugins#getObservableExecutionHook

public RxJavaObservableExecutionHook getObservableExecutionHook() {
    if (observableExecutionHook.get() == null) {
        // 从系统配置文件中查找一个RxJavaObservableExecutionHook的实现类
        Object impl = getPluginImplementationViaProperty(RxJavaObservableExecutionHook.class, System.getProperties());
        // impl = null
        if (impl == null) {
            // 没有找到就使用这个默认的RxJavaObservableExecutionHookDefault实现类
            observableExecutionHook.compareAndSet(null, RxJavaObservableExecutionHookDefault.getInstance());
            // we don't return from here but call get() again in case of thread-race so the winner will always get returned
        } else {
            // we received an implementation from the system property so use it
            observableExecutionHook.compareAndSet(null, (RxJavaObservableExecutionHook) impl);
        }
    }
    return observableExecutionHook.get();
}

rx.plugins.RxJavaObservableExecutionHookDefault

rx.Observable#subscribe(rx.Observer<? super T>)

public final Subscription subscribe(final Observer<? super T> observer) {
    // 使用ObserverSubscriber对observer进行包装
    return subscribe(new ObserverSubscriber<T>(observer));
}

rx.Observable#subscribe(rx.Subscriber<? super T>)

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

rx.Observable#subscribe(rx.Subscriber<? super T>, rx.Observable<T>)

static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
    
    // new Subscriber so onStart it
    subscriber.onStart();
    
    // 保证线程安全
    if (!(subscriber instanceof SafeSubscriber)) {
        // assign to `observer` so we return the protected version
        subscriber = new SafeSubscriber<T>(subscriber);
    }

    try {
        // 调用先RxJavaHooks的onObservableStart的call方法，然后再调用我们在activity中定义的onSubscribe的call方法
        // 这里其实就是调用了开始订阅
        RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);
        return RxJavaHooks.onObservableReturn(subscriber);
    } 
}

rx.plugins.RxJavaHooks#onObservableStart

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

rx.plugins.RxJavaHooks#onObservableReturn

public static Subscription onObservableReturn(Subscription subscription) {
    Func1<Subscription, Subscription> f = onObservableReturn;
    if (f != null) {
        return f.call(subscription);
    }
    return subscription;
}

ObserverSubscriber类

public final class ObserverSubscriber<T> extends Subscriber<T> {
    final Observer<? super T> observer;

    public ObserverSubscriber(Observer<? super T> observer) {
        this.observer = observer;
    }

    @Override
    public void onNext(T t) {
        // 调用observer方法
        observer.onNext(t);
    }

    @Override
    public void onError(Throwable e) {
        observer.onError(e);
    }

    @Override
    public void onCompleted() {
        // 调用observer方法
        observer.onCompleted();
    }
}   

源码阅读总结

Subscription关联观察者和订阅者

static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
    try {
        // 调用先RxJavaHooks的onObservableStart的call方法，然后再调用我们在activity中定义的onSubscribe的call方法
        // 这里其实就是调用了开始订阅
        RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);
        return RxJavaHooks.onObservableReturn(subscriber);
    } 
}

observable.onSubscribe执行

RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);

课堂总结

Observable

1. 观察得到的-被观察者
2. 通过Observable创建一个可观察的序列(create方法)
3. 通过subscribe去注册一个观察者

Observer

1. 用于接收数据-观察者
2. 作为Observable的subsceibe方法的参数

Subscription

1. 订阅，用于描述被观察者和观察者之间的关系
2. 用于取消订阅和获取当前订阅状态

OnSubscribe

1. 当订阅时会触发此接口的调用
2. 在Observable内部，实际作用是向订阅者发射数据

Subscribe

1. 实现了Observer和Subscription
2. 只有自己才能阻止自己