从0实现RxJava

╰+哭是因爲堅強的太久メ 2022-12-17 11:55 113阅读 0赞

参考：[https://mp.weixin.qq.com/s/hiCO5\_0Rc5sXeis13CZrYA][https_mp.weixin.qq.com_s_hiCO5_0Rc5sXeis13CZrYA]  
先做一个最简单RxJava:

/**
     * 观察者
     */
    public interface MyObserver<T> {
    
        void onSubscribe();
        void onNext(T t);
        void onError(Throwable e);
        void onComplete();
    }

//真正的被观察者
    public interface MyObservableOnSubscrible<T> {
    
        void setObserver(MyObserver<T> observer); //设置下游
    }

/**
     * 被观察者
     */
    public class MyObservable<T> {
    
        //上游对象
        private MyObservableOnSubscrible<T> source;
    
        //构造方法，用于接收上游对象
        public MyObservable(MyObservableOnSubscrible<T> source) {
            this.source = source;
        }
    
        //静态方法创建一个真正的被观察者
        public static <T> MyObservable create(MyObservableOnSubscrible<T> source) {
            return new MyObservable(source);
        }
    
        //这里接收一个下游对象，
        public void setObserver(MyObserver<T> observer) {
            source.setObserver(observer);
        } //设置下游
    }

上面三个类就最基本的RxJava类，下面做个简单的使用

MyObservable.create(new MyObservableOnSubscrible<Integer>() {
                @Override
                public void setObserver(MyObserver<Integer> observer) {
                    Log.i(TAG, "上游发送数据:10");
                    observer.onNext(10);
                }
            }).setObserver(new MyObserver<Integer>() {
                @Override
                public void onSubscribe() {
                    Log.i(TAG, "onSubscribe");
                }
    
                @Override
                public void onNext(Integer o) {
                    Log.i(TAG, "下游接收到数据:" + o);
                }
    
                @Override
                public void onError(Throwable e) {
                    Log.i(TAG, "onError");
                }
    
                @Override
                public void onComplete() {
                    Log.i(TAG, "onComplete");
                }
            });

输出结果：

com.example.test  上游发送数据:10
    com.example.test  下游接收到数据:10

是不是跟RxJava 的链式调用非常像，不过就是变量名有点丑，人家的是subscribe方法名，我是setObserver方法名，不过不影响使用。我们后期再把它改过来，现在先这样为了能更好的理解。  
分析一下，发现这里使用了装饰器模式，MyObserver是观察者，MyObservableOnSubscrible是实际上的被观察者，而MyObservable就是装饰类，它持有MyObservableOnSubscrible对象。

**自定义Map操作符**  
显而易见的是map操作符会作用与上游和下游之间，也就是他们的中间。其实这也正是装饰者模式要做的事情，即增强对象。

[https_mp.weixin.qq.com_s_hiCO5_0Rc5sXeis13CZrYA]: https://mp.weixin.qq.com/s/hiCO5_0Rc5sXeis13CZrYA