Rxjava原理(一) -- 响应式编程核心思想的实现

1.Rxjava的使用

​ 关于Rxjava的使用，网上有太多太多，我这里把所有操作符汇总出来，大家有需要了解的自行查找用法，当然我也写了一点点使用方法，在文后的github链接中，这篇文章主要是想写写自己在学习Rxjava中做的笔记，希望能理解得更深刻。

详细的使用方法文档传送门：git@github.com:oujie123/UnderstandingOfRxJava.git

2.Rxjava思想

​ 任何需求只有一个起点和一个终点，起点开始流向一个”事件”，”事件”经过处理之后最终流向终点，期间事件的处理可以分为很多步骤，处理的过程和终点只关心上一个过程的结果。在订阅之前的流程称之为上游，在订阅之后的流程称之为下游。

3.Rxjava的hook点

​ 在看源码之前，多看几个api，细心的朋友都会看到在每个API返回的之后都会有RxJavaPlugins.onAssembly()方法。他的作用是什么呢？

//io.reactivex.plugins.RxJavaPlugins类
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;
}

默认情况下onObservableAssembly为null，导致if语句不起作用。那么Rxjava设计这个有什么用呢？他的主要用途就是用于hook每一次调用。注意，这个方法会在每一个API最后中调用，意味着每一个Rxjava的方法都会经历这个方法。所以就可以通过这个方法来hook开发者的每一次调用。那么使用方法是怎样的呢？、

可以通过一下方法设置一个处理方法

public static void setOnObservableAssembly(@Nullable Function<? super Observable, ? extends Observable> onObservableAssembly) {
    if (lockdown) {
        throw new IllegalStateException("Plugins can't be changed anymore");
    }
    RxJavaPlugins.onObservableAssembly = onObservableAssembly;
}

// apply中可以加入自己想实现的代码
RxJavaPlugins.setOnObservableAssembly(new Function<io.reactivex.Observable, io.reactivex.Observable>() {
            @Override
            public io.reactivex.Observable apply(io.reactivex.Observable observable) throws Exception {
                Log.i("jack","api of rxjava is invoked ！");
                return observable;
            }
        });

4.Rxjava核心实现

首先列出基本代码，其他的操作符都是一样的

//返回Observable<String>  真实的是ObservaleCreate<String>()
Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> e) throws Exception {
        e.onNext("JackOu");
    }
})

//ObservaleCreate.map
.map(new Function<String, Bitmap>() {
            @Override
            public Bitmap apply(String s) throws Exception {
                return null;
            }
        })

 //ObservableMap.subscribe()
 .subscribe(new Observer<Bitmap>() {
     @Override
     public void onSubscribe(Disposable d) {
     }

     @Override
     public void onNext(Bitmap bitmap) {
     }

     @Override
     public void onError(Throwable e) {
     }

     @Override
     public void onComplete() {
     }
 });

分析分为3步：

• 创建Observer，就是new一个接口 自定义观察者
• 创建Observable
• subscribe订阅过程

4.1 创建观察者

public interface Observer<T> {

    // 当订阅成功后回调
    void onSubscribe(@NonNull Disposable d);

    // 开发者调用onNext回调
    void onNext(@NonNull T t);

   // 通知观察者发生异常时回调
    void onError(@NonNull Throwable e);

    // 事件处理结束回调
    void onComplete();

}

观察者的创建就是创建一个回调实现，用于接收上游传下来的事件结果。

4.2 创建Observable

调用流程如下：

// 第一步调用create方法，传入自定义source
Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> e) throws Exception {
        e.onNext("novel1");
    }
})
    
 //创建ObservableCreate封装自定义source
 public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        ObjectHelper.requireNonNull(source, "source is null");
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }   
//最终用ObservableCreate包裹了自定义source（ObservableOnSubscribe接口）

4.3 subscribe订阅过程

// 1.当subscribe()一被调用，Observable的subscribe()方法会被调用
public final void subscribe(Observer<? super T> observer) {
    ......
    try {
       ......
		// 调入子类
        subscribeActual(observer);
    } catch (NullPointerException e) { // NOPMD
    ......
}
    
// 2.调用子类ObservaleCreate的subscribeActual方法
protected void subscribeActual(Observer<? super T> observer) {
    	// 创建发射器包裹自定义的观察者
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
    	// 调用观察者的onSubscribe订阅回调
        observer.onSubscribe(parent);

        try {
            // 上述创建的自定义source，即ObservableOnSubscribe
            // 调用ObservableOnSubscribe的subscribe传入自定义观察者
            // 开发者调用发射器的onNext()就会调到自定义观察者的onNext()中
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }

4.4 原理图

5.标准观察者模式和Rxjava的变异观察者模式有哪些区别

​ 在标准的观察者模式，是一个“被观察者”，多个“观察者”，当被观察者有事件通知的时候，所有观察者都会收到；在标准的观察者设计模式中：当发出通知改变时，会遍历Observable里面的容器，此容器里面有10个Observer，就会通知10个Observer
​ rxjava的观察者模式是多个“被观察者”，一个“观察者”，需要起点和终点订阅一次之后，但发出改变通知，观察者才能看到。在RxJava观察者设计模式中：分发事件时，会拿到发射器，通过发射器关联到我们自定义的Observer，发射器调用到我们自定义的Observer

6.在创建类操作符中create和just的区别

​ just内部封装了一个run方法，自动执行了，observer.onNext()

​ create需要程序员去调用createEmitter.onNext（）