从0实现RxJava

RxJava"/>

从0实现RxJava

参考：
先做一个最简单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>() {@Overridepublic void setObserver(MyObserver<Integer> observer) {Log.i(TAG, "上游发送数据:10");observer.onNext(10);}}).setObserver(new MyObserver<Integer>() {@Overridepublic void onSubscribe() {Log.i(TAG, "onSubscribe");}@Overridepublic void onNext(Integer o) {Log.i(TAG, "下游接收到数据:" + o);}@Overridepublic void onError(Throwable e) {Log.i(TAG, "onError");}@Overridepublic void onComplete() {Log.i(TAG, "onComplete");}});

输出结果：

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

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

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