【移动应用开发技术】还不会Rxjava响应式编程框架设计先从这篇文章入手

【移动应用开发技术】还不会Rxjava响应式编程框架设计，先从这篇文章入手

Rxjava在GitHub主页上的自我介绍是"alibraryforcomposingasynchronousandevent-basedprogramsusingobservablesequencesfortheJavaVM"（一个在JavaVM上使用可观测的序列来组成异步的、基于事件的程序的库）。通俗来说，Rxjava是一个采用了观察者模式设计处理异步的框架。链式调用设计让代码优雅易读。举个例子：Observable<String>observable=Observable.create(newObservableOnSubscribe<String>(){

@Override

publicvoidsubscribe(ObservableEmitter<String>e)throwsException{

e.onNext("a");

}

});

observable.subscribe(newObserver<String>(){

@Override

publicvoidonSubscribe(Disposabled){

}

@Override

publicvoidonNext(Strings){

}

@Override

publicvoidonError(Throwablee){

}

@Override

publicvoidonComplete(){

}

});这是Rxjava2最简单的用法：1.创建一个Observable，重写subscribe方法，这里主要处理被观察的事件。2.订阅这个Observable，事件会回调observer的方法，我们可以对事件做响应的处理2.1.创建Observable：创建Observable用的是Observable.create(ObservableOnSubscribe<T>source)方法。这个方法的参数是ObservableOnSubscribe：publicinterfaceObservableOnSubscribe<T>{

/**

*CalledforeachObserverthatsubscribes.

*@paramethesafeemitterinstance,nevernull

*@throwsExceptiononerror

*/

voidsubscribe(@NonNullObservableEmitter<T>e)throwsException;

}ObservableOnSubscribe是一个接口，唯一的方法是subscribe，参数是ObservableEmitter<T>e。ObservableEmitter是一个继承了Emitter的接口，接口Emitter里定义了onNext、onError、onComplete等方法，和Observer（观察者）的方法相对应。publicinterfaceEmitter<T>{

/**

*Signalanormalvalue.

*@paramvaluethevaluetosignal,notnull

*/

voidonNext(@NonNullTvalue);

/**

*SignalaThrowableexception.

*@paramerrortheThrowabletosignal,notnull

*/

voidonError(@NonNullThrowableerror);

/**

*Signalacompletion.

*/

voidonComplete();

}ObservableEmitter对接口Emitter进行扩展，增加了setDisposable、setCancellable等方法基本参数了解了，现在看看create方法里面做了什么，代码如下：publicstatic<T>Observable<T>create(ObservableOnSubscribe<T>source){

returnRxJavaPlugins.onAssembly(newObservableCreate<T>(source));

}

调用了RxJavaPlugins的onAssembly方法。又有一个新参数ObservableCreate<T>(source)，我们看看它是什么：

finalclassObservableCreate<T>extendsObservable<T>{

publicObservableCreate(ObservableOnSubscribe<T>source){

this.source=source;

}

}继承了Observable，所以也是个被观察对象，在构造函数中我们看到我们new的ObservableOnSubscribe对象，被存在了ObservableCreate的source里面那我们继续看看onAssembly方法做什么：publicstatic<T>Observable<T>onAssembly(@NonNullObservable<T>source){

Function<?superObservable,?extendsObservable>f=onObservableAssembly;

if(f!=null){

returnapply(f,source);

}

returnsource;

}一个Hook方法。onObservableAssembly是一个静态变量，我们没有设置，默认为空，所以直接返回source对象。也就是说，Observable的create方法其实就是把我们ObservableOnSubscribe对象，存储在ObservableCreate对象的source里面，然后返回ObservableCreate对象。我们知道ObservableCreate是继承Observable的，所以创建了ObservableCreate对象，我们的Observable也就创建完了。2.2订阅事件（被观察者）订阅被观察者的操作是observable.subscribe(newObserver<String>())。这个操作符其实是个“被动”，就是事件被观察者观察。因为subscribe方法里的参数Observer才是观察者。我们也会在Observer里的各个会调方法里接收到事件相关的返回值。我们看看subscribe方法的源码：publicfinalvoidsubscribe(Observer<?superT>observer){

try{

subscribeActual(observer);

}catch(NullPointerExceptione){//NOPMD

throwe;

}catch(Throwablee){

RxJavaPlugins.onError(e);

}

}看代码我们知道最主要调用的方法是：subscribeActual(observer);，这个方法是Observable里的抽象方法，而此时我们的Observable是一个ObservableCreate对象(前面create方法返回的对象)。所以我们去看一下ObservableCreate里面是如何重写这个方法的。代码如下：protectedvoidsubscribeActual(Observer<?superT>observer){

CreateEmitter<T>parent=newCreateEmitter<T>(observer);

observer.onSubscribe(parent);

try{

source.subscribe(parent);

}catch(Throwableex){

Exceptions.throwIfFatal(ex);

parent.onError(ex);

}

我们一看到这个方法主要做了三件事：①创建一个CreateEmitter对象parent。②把parent传给source的subscribe方法。上面我们知道source就是刚才存的ObservableOnSubscribe对象，subscribe也就是我们重写的方法：@Override

publicvoidsubscribe(ObservableEmitter<String>e)throwsException{

e.onNext("a");

}所以我们在这个方法里就能收到一个CreateEmmiter，通过CreateEmitter可以回调相应的方法。CreateEmitter是实现ObservableEmitter接口，我们看看它内部实现，如：onNext源码如下：@Override

publicvoidonNext(Tt){

observer.onNext(t);

}也就是说，当我们在ObservableOnSubscribe的subscribe方法里调用ObservableEmitter的onNext方法的时候，它里面会调用observer的onNext。于是通过这样的传递，我们就能在observer里响应的回调方法里收到事件的相关状态。至此一个简单Rxjava流式传递原理已经讲完了，总结流程如下：使用Observbable.create方法，产生一个ObservableCreate对象，对象里存着ObservableOnSubscribe对象source。调用ObservableCreate.subscribe方法，实际调用的是subscribeActual方法，传入一个Observer对象。subscribeActual方法中创建一个CreateEmmiter对象，调用source.subscribe方法，传入CreateEmmiter对象。于是我们