再一次对RxJava的理解和感悟，手写简单实现

RxJava的理解和感悟

最近又看了下RxJava 感觉又领悟了些东西，于是想写下来分享一下

首先再RxJava中经常看到这两个角色observable和observer，一看这两个单词就想到了观察者模式，这里observable是被观察者一般用于获取数据，observer是观察者，用于监听然后处理被观察者的产生的数据，大体上RxJava就是这样的流程，但是他的写法和传统设计模式中的写法不同

RxJava中的观察者模式

RxJava中的observable在注册一个observer时，并不是用一个数组去存储，而是通过注册方法也就是

subscribe(@NonNull Observer<? super T> observer) 方法，

把参数observer直接传给调用者也就是我们程序员，然后我们调用observer中处理数据的方法，也就是onNext（）方法获取到observable中产生的数据。下面我就自己去实现一下这个操作

观察者

/**
 * 观察者接口
 */
public interface RXObserver {
    public void  onNext(String s);
}

 

/**
 * 被观察者接口
 */
public interface RXObservable {
    public void addObserver(RXObserver observer);
}

/**
 * 被观察者的抽象类
 */
public abstract  class AbsObservable implements RXObservable {
    @Override
    public void addObserver(RXObserver observer) {
        addObserverTrue(observer);
    }

    public abstract void addObserverTrue(RXObserver observer);

   
}

 

 

/**
 * 具体的某个被观察者
 */
public class MyObservable extends AbsObservable {


    public MyObservable() {
       
    }


    /**
     * 添加注册观察者 来处理我的数据
     * @param observer 观察者
     */
    @Override
    public void addObserverTrue(RXObserver observer) {
        String data="1111";// 我生成一个数据 给observer处理
       observer.onNext(data);//处理数据

    }


}

这里AbsObservable是一个抽象的被观察类，里面有一个抽象方法addObserverTrue（）就是RxJava中的subscribe方法用来添加一个观察者给被观察者。但是这个是一个最简单的流程，于是要再接着改造成RxJava那样

首先，处理产生数据和处理数据肯定不是固定，是要给调用者实现，于是这部分要通过一接口让调用者实现里面的。那这个接口中的方法怎么定义呢。前面说了这个接口是要产生数据和处理数据的。产生数据很简单，你可以声明一个数据，或者调用网络工具类或者其他的方法得到数据，那处理数据呢？处理是需要观察者observer处理，那这个observer怎么来呢。肯定是我们注册进来的，所以这个接口要把注册进来的观察者再回调给调用者，于是就变成了这样

首先接口

public interface ObserverCallback {
    public void callBack(RXObserver observer);//把observer再次回调出去
}

 

然后之前的观察者就变成了这样了

/**
 * 被观察者
 */
public class MyObservable extends AbsObservable {
    ObserverCallback callback;

    public MyObservable(ObserverCallback callback) {
        this.callback = callback;
    }
    


    /**
     * 添加注册观察者 来处理我的数据
     * @param observer
     */
    @Override
    public void addObserverTrue(RXObserver observer) {
//        String data="1111";// 我生成一个数据 给observer处理
//       observer.onNext(data);//
        callback.callBack(observer);//回传出去，让调用者处理

    }


}

这里接口ObserverCallback就是负责把observer回调出去，就是把我们传进来的观察者在回调出去。。。。。然后操纵被观察者，只是绕了一个弯子

这样调用的时候就变成这样了，有点像RxJava了

public void toFun(){
    new MyObservable(new ObserverCallback() {
        @Override
        public void callBack(RXObserver observer) {
            observer.onNext("111");
        }
    }).addObserver(new RXObserver() {
        @Override
        public void onNext(String s) {

        }
    });

}

 

但是RxJava不是直接把observer传递出来而是通过

CreateEmitter

类包装了一下，然后把CreateEmitter回调出去，他和观察者可以理解为是一样的东西

好了这就是我对RxJava的一个简单的理解，下回讲一个RxJava中的一个操作符Map的实现，加深理解和事件的传递

 

最后

安利一下对 RxJava源码解读的文章（共三个部分） 通俗易懂

https://juejin.im/post/5a521d68f265da3e4e25750e

https://juejin.im/post/5a6751af6fb9a01cb2571794

https://juejin.im/post/5b15486e6fb9a01e657a4bcc