在Android上打造属于自己的Rxjava框架(一)基础篇

前言 .最近发现Rxjava比较火，所以去了解了一下，发现确实挺有意思的，于是乎迫不及待的下载了源码下来研究一番。然后就迫不及待的写了一份类似的，与RxJava还是有很大不同的。一是接触时间太短，匆匆看了几行源码就开始写了，二是平时确实不用JS，那种函数式响应式编程的感觉很薄弱吧。所以还是按照平时我在android编码里的习惯设计的这个小框架。

1.基础:观察者模式,都是基于观察者模式那是肯定的，这个也是框架的核心。

Rxjava据我短时间的理解:在原理上就是:事件的发布(包括那些“匿名”了的事件)->数据打包变换截取缓存等操作(Operator)->事件的订阅,当然还包括中间的调度器。在编程习惯上是:保持链式调用,配合lambda表达式获得"赏心悦目"的编码效果。

(1).基础接口:OnPublish,OnObserver。两个核心接口，几乎所有的操作都围绕这两个展开，和Rxjava几乎一致。

public interface OnPublisher<T> {
public void call(OnObserver<T> observer);
}

public interface OnObserver<T> {
public void onSuccess(T t);
public void onError(Throwable throwable);
public void onFinished();
}

发布者Publish执行call方法，在call方法中调用事件订阅者中的三种状态回调。

(2).主体类Publisher的设计，对应RxJava的Observable。个人感觉方法还是抽象在接口里为好，虽然RxJava没有这么做。链式调用的核心就是调用的方法返回Publisher类对象以达到持续调用Publisher中方法的目的。而使用接口，因为不同的接口对应特定的方法集合，只要结合方法的返回类型就能控制链式调用的顺序问题，当然这些都是后话，暂时用不到这些。

public interface IPublisher<T> {
public IPublisher<T> create(OnPublisher<T> onPublisher);
public IPublisher<T> create(T[] ts);
public IPublisher<T> create(Iterator<T> ts);
public IPublisher<T> name(String t);
public IPublisher<T> save();
public void destroy();
public <I,O> IPublisher<T> change(Func1<I,O> change,Change changeProxy);
public IPublisher<T> with(T t);
public IPublisher<T> with(T[] ts);
public IPublisher<T> with(Iterator<T> ts);
public IPublisher<T> bind(OnObserver<T> observer);
public IPublisher<T> bind(Action1<T> action1);
public IPublisher<T> filter(OnFilter<T> filter);
public IPublisher<T> filted(OnFilter<T> filter);
public IPublisher<T> just(T t);
public <I,O> IPublisher<T> map(Func1<I,O> map);
public <I,O> IPublisher<T> maped(Func1<I,O> maped);
public IPublisher<T> mapOn(RunContextType contextType);
public IPublisher<T> remove(OnObserver<T> observer);
public IPublisher<T> reMap(Func1<T,IPublisher<T>> publisher);
public IPublisher<T> link();
public IPublisher<T> clear();
public IPublisher<T> post();
}

接口方法有点多，但是可以发现返回类型都是IPublisher，知道destory返回void才表示链式调用的终点。另外，这个接口的范型其实是有些问题的，不利于后续的map等变换操作，后续是会更改的。不过范型也仅仅只是源码级的约束而已，并不会影响RunTime的结果。

下面是一个最简单的Publisher，也是原型。仅仅实现了 发布——多订阅

public class Publisher<T> {

private OnPublisher<T> onPublisher;
private Vector<OnObserver<T>> onObservers;

public static  <T> Publisher<T> getInstance(){
return new Publisher<T>();
}

private Publisher() {
onObservers = new Vector<OnObserver<T>>();
}

public Publisher<T> create(OnPublisher<T> publisher){
onPublisher = publisher;
return this;
}
public Publisher<T> bind(OnObserver<T> observer){
onObservers.add(observer);
return this;
}
public Publisher<T> post(){
onPublisher.call(new Observer<T>() {
@Override
public void onSuccess(T t) {
for (OnObserver<T> observer:onObservers)
observer.onSuccess(t);
}

@Override
public void onError(Throwable throwable) {
for (OnObserver<T> observer:onObservers)
observer.onError(throwable);
}

@Override
public void onFinished() {
for (OnObserver<T> observer:onObservers)
observer.onFinished();
}
});
return this;
}
}

这个简单版本并没有继承上面的IPubulisher接口。事件订阅完成之后由post方法统一触发。在call方法中的参数其实是OnObserver的匿名代理，用来包装遍历多个OnObserver。

可以看出从这里开始就与RxJava有些不同了，Rxjava是事件订阅的时候就执行触发事件，可以说是订阅一下触发一次。其实这里的想法出现了分歧，Rxjava其实在事件消费的时机上与传统的观察者模式有所不同。传统观察者模式都是在所有事件订阅完成之后，发出信号而后所有订阅者统一收到同一份事件，而Rxjava则是订阅即触发单独的事件消费。

说了这么多其实事件消费的策略对于安卓来说个人认为还是传统的观察者模式更常用一些。当然我的想法归我的想法如有差错还请大手们指正。当然事件消费的时机在设计上其实可以做到两者兼有，虽然我暂时只做了传统的策略，但是从设计上来说并不影响以后的扩展。

(3).Rxjava为了保持链式调用从而让Observable去subscribe Observer，一开始我也看着很奇怪，这就像事件的发布者订阅了事件的订阅者一般。所以我把subscribe换成了bind自我感觉更合适一些。当然稍稍扩展一下，Observer还是能subscribe Publisher的，如果开发者不介意打破链式的话。

public class Subscriber<T> implements ISubscriber<T>,OnObserver<T> {

private OnObserver<T> onObserver;

public static <T> Subscriber getInstance(OnObserver<T> onObserver){
return new Subscriber<T>(onObserver);
}

public Subscriber(OnObserver<T> onObserver) {
this.onObserver = onObserver;
}

@Override
public ISubscriber<T> subscribe(IPublisher<T> pub) {
pub.bind(this);
return this;
}

@Override
public void onSuccess(T t) {
onObserver.onSuccess(t);
}

@Override
public void onError(Throwable throwable) {
onObserver.onError(throwable);
}

@Override
public void onFinished() {
onObserver.onFinished();
}

public OnObserver<T> getOnObserver() {
return onObserver;
}

public void setOnObserver(OnObserver<T> onObserver) {
this.onObserver = onObserver;
}

}

其实也就是代理模式，在代理类中加了一个subscribe方法，将他命名为Subscriber。

第一篇的结尾:预告一下，第二篇将实现调度器，让事件在不同线程中发布和消费，提示一下与Rxjava的实现有很大不同哦，第三篇实现变幻即Rxjava的操作符。

源码:介绍一下自己的写的Android快速开发框架SwiftFrameWork，GitHub