RxJava源码浅析（一）

源码地址：RxJava

1. 功能介绍

1.1 RxJava

RxJava is a Java VM implementation of ReactiveX (Reactive Extensions): a library for composing asynchronous and event-based programs by using observable sequences.
通过观察者模式，实现异步和基于事件编程

1.2 特点

轻量级，专注于Observable被观察的抽象和一系列的方法

2. 流程图

1. 创建一个Observable对象（被观察者）
2. 创建一个Subscriber对象（观察者）
3. 订阅观察者
4. 触发观察者回调

实现DEMO：

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

2.1 创建Observable被观察者

通过create()方法来创建一个Observable对象

static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
public static <T> Observable<T> create(OnSubscribe<T> f) {
    return new Observable<T>(hook.onCreate(f));
}
public static <S, T> Observable<T> create(SyncOnSubscribe<S, T> syncOnSubscribe) {
    return new Observable<T>(hook.onCreate(syncOnSubscribe));
}
public static <S, T> Observable<T> create(AsyncOnSubscribe<S, T> asyncOnSubscribe) {
    return new Observable<T>(hook.onCreate(asyncOnSubscribe));
}
public interface OnSubscribe<T> extends Action1<Subscriber<? super T>> {
    // cover for generics insanity
}

3个方法都试执行hook的create()方法

public <T> OnSubscribe<T> onCreate(OnSubscribe<T> f) {
    return f;
}

直接返回来一个OnSubscribe对象，实现call()方法

public interface Action1<T> extends Action {
    void call(T t);
}

2.2 创建Subscriber（观察者）

public abstract class Subscriber<T> implements Observer<T>, Subscription 

直接new出来后实现未实现的方法

new Subscriber<String>() {
    @Override
    public void onCompleted() {
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onNext(String s) {
    }
});

onNext()普通触发事件
onCompleted()事件队列完结，当触发事件队列完成后触发
onError()事件队列异常，在事件处理过程中出异常时触发，事件终止，不会再有事件触发
onCompleted()和onError()只会触发一个

2.3 订阅观察者

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();
    hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
    return hook.onSubscribeReturn(subscriber);
｝

hook的onSubscribeStart()返回的就是一个Action1对象再执行call()方法

public <T> OnSubscribe<T> onSubscribeStart(Observable<? extends T> observableInstance, final OnSubscribe<T> onSubscribe) {
    // pass through by default    
    return onSubscribe;
}

最后来说说hook，hook是一个 proxy 对象, 仅仅用作调试的时候可以插入一些测试代码

2.4 触发观察者回调

调用subscriber对象的 onNext()，onComplete() 等方法

2.5 总结

Observable创建了OnSubscribe对象，该对象重载了call() 方法，触发通过 subscribe() 方法绑定了Subscriber对象，该对象重载了 onNext()，onComplete()，onError() 方法

未完待续