ReactiveX框架(基于RxJava)实现原理浅析
来自： http://blog.kifile.com/reactivex/2015/12/07/rx_introduce.html
前言
先简单介绍一下 ReactiveX.
ReactiveX 并不特指某种编程语言,他应该算是一种编程思维,反应式编程.
反应式编程的核心在于,当触发特定行为逻辑后(对于ReactiveX而言,就是调用了 subscribe 指令),根据进行指定操作,并根据操作结果执行特定操作. 这种编程思维特别适合用于交互式软件上,例如Android,iOS,通常用户触发某个条件(比如说点击操作)后,我们需要根据用户的操作行为, 可能接下来要执行一系列操作,最后再根据操作结果在ui界面上呈现给用户.而ReactiveX 为我们提供了这种交互流程的封装.
下面以RxJava为例,分析一下ReactiveX的框架实现原理.
RxJava执行流程
首先奉上一个最简单的 ReactiveX 的代码实现.
1 Observable.create(new Observable.OnSubscribe<String>() { 2 @Override 3 public void call(Subscriber<? super String> subscriber) { 4 subscriber.onNext("Sample"); 5 subscriber.onCompleted(); 6 } 7 }).subscribe(new Observer<String>() { 8 @Override 9 public void onCompleted() { 10 System.out.println("Complete"); 11 } 12 13 @Override 14 public void onError(Throwable e) { 15 e.printStackTrace(); 16 } 17 18 @Override 19 public void onNext(String s) { 20 System.out.println("Get:" + s); 21 } 22 }); 运行上面的代码,我们可以看到以下输出.
Get:Sample Complete
代码执行完毕,让我们看看整个流程的实现逻辑.
1 private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) { 2 ... 3 subscriber.onStart(); 4 ... 5 try { 6 hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber); 7 return hook.onSubscribeReturn(subscriber); 8 } catch (Throwable e) { 9 Exceptions.throwIfFatal(e); 10 try { 11 subscriber.onError(hook.onSubscribeError(e)); 12 } catch (Throwable e2) { 13 Exceptions.throwIfFatal(e2); 14 // if this happens it means the onError itself failed (perhaps an invalid function implementation) 15 // so we are unable to propagate the error correctly and will just throw 16 RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2); 17 // TODO could the hook be the cause of the error in the on error handling. 18 hook.onSubscribeError(r); 19 // TODO why aren't we throwing the hook's return value. 20 throw r; 21 } 22 return Subscriptions.unsubscribed(); 23 } 24 }
在上面的代码里,会发现一个 hook 对象,这是个什么鬼?
追踪一下,发现原来他是一个RxJava每个方法都会返回一个Observable对象ExecutionHook对象,类图如下:
可以看出,RxJavaObservableExecutionHook中针对RxJava的subscribe流程进行注入,方便自己更改相关逻辑, 当然对于默认的RxJavaObservableExecutionHook,你会发现他并没有做任何处理,如果你想自己实现可以调用 RxJavaPlugin.getInstance() 设置自定义Hook.
看完上面的例子,给人感觉挺简单地啊,而且比较类似Android中的AsyncTask,都是属于执行任务后进行回调,那他相比AsyncTask有什么优势吗?
Operator
虽然从上面的例子中,看起来RxJava的确其貌不扬,但是ReactiveX也不止这点技法.
为了扩展ReactiveX的相关属性,在 RxJava 中使用代理模式实现了很多有用的逻辑,例如类型转换,遍历数据个数限制,定时响应等额外特性.
这些逻辑被称为操作(Operator),每一个Operator都继承了Func1(也就是内部会有一个call()方法),RxJava框架会调用lift方法将Operator包装成为Observable:
1 public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) { 2 return new Observable<R>(new OnSubscribe<R>() { 3 @Override 4 public void call(Subscriber<? super R> o) { 5 try { 6 Subscriber<? super T> st = hook.onLift(operator).call(o); 7 try { 8 // new Subscriber created and being subscribed with so 'onStart' it 9 st.onStart(); 10 onSubscribe.call(st); 11 } catch (Throwable e) { 12 // localized capture of errors rather than it skipping all operators 13 // and ending up in the try/catch of the subscribe method which then 14 // prevents onErrorResumeNext and other similar approaches to error handling 15 Exceptions.throwIfFatal(e); 16 st.onError(e); 17 } 18 } catch (Throwable e) { 19 Exceptions.throwIfFatal(e); 20 // if the lift function failed all we can do is pass the error to the final Subscriber 21 // as we don't have the operator available to us 22 o.onError(e); 23 } 24 } 25 }); 26 }
如果想了解更多详细的操作信息,可以点击这里: Operators
其实这里有个问题,既然大部分的Operator都需要这样封装,为什么不直接让Operator对象继承OnSubscribe对象,进而减少方法调用层级?
如果是为了防止OnSubscribe和Func1的方法重名,那么更改函数名就好了啊? 如果为了进行onLift回调,也可以在新类中增加回调调用位置啊?
同时为了避免代码冗余,对于这些方法,RxJava都使用了构造者模式的一种变体,每个方法都会返回一个Observable对象,保证其能够形成类似下面这样的操作链.
1 Observable.just("Hello", "Operator", "Chain").map(s -> s + " map" ) 2 .buffer(2).take(1).subscribe(new Subscriber<List<String>>() { 3 @Override 4 public void onCompleted() { 5 System.out.println("Complete"); 6 } 7 8 @Override 9 public void onError(Throwable e) { 10 11 } 12 13 @Override 14 public void onNext(List<String> strings) { 15 System.out.println("Get:" + strings); 16 } 17 });
输出结果如下:
Get:[Hello map, Operator map] Complete
可以看到,我们输入了三个String对象,但是只一次输入了两个String,并且两个String 都额外多了一个 map 后缀,当然这也是我想要的结果
使用Single替代Observable
其实在绝大部分的使用场景中,用户触发操作后,对于我们而言返回结果其实只有成功失败两种,而Observable中是有 onNext , onComplete , onError 三种状态,这样看来似乎不太满足需求.
当然ReactiveX中也考虑到了这种情况,它在Observable的基础上衍生出了Single类,这个类的实现机制同Observable近乎相同,只是不过订阅他的不再是 Subscriber 对象,而是 SingleSubscriber .
在 SingleSubscriber 中有 onSuccess 和 onError 两种结果状态,并且只会调用其中一个,恰好满足我们的需求.
使用Scheduler进行多线程调用
按照上面所说,不论是Observable还是Single,其实都是在同一个线程中,不断按照执行逻辑执行代码指令,也就是说他始终是在同一线程中进行执行的.
但是有时候,我们希望能够在异步线程中执行耗时操作,避免ui堵塞,这时候ReactiveX就为我们提供了 Scheduler 类.
Scheduler类其实并不负责异步线程处理,他只负责通过 createWorker() 类创建出一个 Worker 对象,真正负责任务的延时处理.
每一个Scheduler类,都会实现自己的Worker类,用于执行Scheduler任务.
我们可以使用ReactiveX中的 subscribeOn 和 observeOn 两个方法,分别设置获取数据的操作和分发消息的操作的执行Scheduler,从而实现数据的异步处理.
subscribeOn 和 observeOn 其实都是构建一个Operator对象,在call方法里,使用线程执行数据获取和分发操作.
总结
其实ReactiveX就是一个针对观察者模式的扩展,如果忽略掉ReactiveX框架为我们实现各种的Operator,那么它就是一个简单的设计模式而已.
单就这一点而言,Android的AsyncTask和LoaderManager框架是胜过ReactiveX的,因为他针对Android的生命周期做了处理.
但由于ReactiveX中增加了很多Operator,能够很方便的帮助我们对响应式任务进行操作,不论是类型转换还是异步执行.
反而导致看起来ReactiveX中的比AsyncTask要好用,但我觉得如果吧ReactiveX结合到AsyncTask和LoaderManager中应该会更加的完美.