ReactiveX框架(基于RxJava)實現原理淺析

在上面的代碼里,會發現一個 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 }

 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