rxJava2.x源码解析

2022-01-04 21:40:22

一. Rxjava是什么

　　Rxjava在GitHub的介绍是 "A library for composing asynchronous and event-based programs using observable sequences for the Java VM"（一个在 JVM上使用可观测的序列来组成异步的、基于事件的程序的库）。

Rxjava是一个采用了观察者模式设计处理异步的基于事件机制的框架。另外，链式调用、逻辑简单清晰。

看下rxjava无背压模式的简单用法：

 1 Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
 2 
 3             @Override
 4             public void subscribe(ObservableEmitter<String> emitter) throws Exception {
 5                 emitter.onNext("Hello the world");
 6             }
 7         });
 8 
 9         observable.subscribe(new Observer<String>() {
10             @Override
11             public void onSubscribe(Disposable d) {
12 
13             }
14 
15             @Override
16             public void onNext(String s) {
17 
18             }
19 
20             @Override
21             public void one rror(Throwable e) {
22 
23             }
24 
25             @Override
26             public void onComplete() {
27 
28             }
29         });

1. 创建一个Observable，重写subscribe方法，这里主要处理被观察的事件。
2. 订阅这个Observable，事件会回调observer的方法，可以对事件做响应的处理

二. 无背压模式的源码解析

2.1. 创建Observable：

创建Observable用的是Observable.create(ObservableOnSubscribe<T> source)方法。这个方法的参数是ObservableOnSubscribe：

public interface ObservableOnSubscribe<T> {

    /**
     * Called for each Observer that subscribes.
     * @param e the safe emitter instance, never null
     * @throws Exception on error
     */
    void subscribe(@NonNull ObservableEmitter<T> e) throws Exception;
}

ObservableOnSubscribe是一个函数式接口，有唯一的方法subscribe，参数是ObservableEmitter<T> e。ObservableEmitter是一个继承了Emitter的接口，接口Emitter里定义了onNext、onError、onComplete等方法，和Observer（观察者）的方法相对应。

public interface Emitter<T> {

    /**
     * Signal a normal value.
     * @param value the value to signal, not null
     */
    void onNext(@NonNull T value);

    /**
     * Signal a Throwable exception.
     * @param error the Throwable to signal, not null
     */
    void one rror(@NonNull Throwable error);

    /**
     * Signal a completion.
     */
    void onComplete();
}

ObservableEmitter对接口Emitter进行扩展，增加了setDisposable、setCancellable等方法
下面看看create方法：

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }

调用了RxJavaPlugins的onAssembly方法。又有一个新参数ObservableCreate<T>(source)，我们看看它是什么：

final class ObservableCreate<T> extends Observable<T> {

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

}

继承了Observable，是被观察对象，在构造函数中我们看到我们new ObservableOnSubscribe对象，被存在了ObservableCreate source引用里

onAssembly方法：

public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
        Function<? super Observable, ? extends Observable> f = onObservableAssembly;
        if (f != null) {
            return apply(f, source);
        }
        return source;
    }

一个Hook方法。onObservableAssembly是一个静态变量，没有设置默认为空，所以直接返回source对象。也就是说，Observable的create方法其实就是把ObservableOnSubscribe对象存储在ObservableCreate对象的source引用里，然后返回该ObservableCreate对象。
ObservableCreate是继承Observable的，所以创建了ObservableCreate对象，Observable也就创建完了。

2.2 订阅事件（被观察者）

订阅操作是observable.subscribe(new Observer<String>())。这里是被观察者订阅观察者，主要是因为链式调用方便，因为subscribe方法里的参数Observer才是观察者。我们也会在Observer里的各个被调方法里接收到事件相关的返回值。
subscribe订阅方法的源码：

  public final void subscribe(Observer<? super T> observer) {
        try {
            subscribeActual(observer);
        } catch (NullPointerException e) { // NOPMD
            throw e;
        } catch (Throwable e) {
            RxJavaPlugins.onError(e);
        }
    }

实际上调用了subscribeActual(observer);，这个方法是Observable里的方法，而此时的Observable是一个ObservableCreate对象。ObservableCreate里面的subscribeActual如下：

protected void subscribeActual(Observer<? super T> observer) {
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        observer.onSubscribe(parent);

        try {
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }

方法主要做了三件事：

1. 创建一个CreateEmitter对象parent;
2. 把parent传给source的subscribe方法，也就是ObservableOnSubscribe对象的subscribe方法：

@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
     e.onNext("a");
}

所以在这个方法里就能收到一个CreateEmmiter，通过CreateEmitter可以回调相应的方法。CreateEmitter是实现ObservableEmitter接口，内部onNext源码如下：

@Override
public void onNext(T t) {
    observer.onNext(t);
}

也就是说，当我们在ObservableOnSubscribe的subscribe方法里调用ObservableEmitter的onNext方法的时候，它里面会调用observer的onNext。于是通过这样的传递就能在observer的回调方法里收到。

总结无背压Rxjava（observable/observer）：

1、使用Observbable.create方法，产生一个ObservableCreate对象，对象里存着ObservableOnSubscribe对象source。

2、调用ObservableCreate.subscribe方法，实际调用的是subscribeActual方法，传入一个Observer对象。

3、subscribeActual方法中创建一个CreateEmmiter对象，调用source.subscribe方法，传入CreateEmmiter对象。

4、于是我们在ObservableOnSubscribe中就接收到了一个CreateEmmiter，CreateEmmiter是ObservableEmmiter的子类。我们可以在这里调用CreateEmmiter的方法进行事件回调。

5、调用CreateEmmiter方法，实际上会调用Observer的响应的方法。也就是CreateEmmiter把事件状态传递给观察者。

码农公寓

一. Rxjava是什么

二. 无背压模式的源码解析

2.1. 创建Observable：

2.2 订阅事件（被观察者）

总结无背压Rxjava（observable/observer）：

相关文章