RxJava 簡(jiǎn)單來(lái)說(shuō) , 是一個(gè)很靈活切換線程的褲子 .
- 簡(jiǎn)單試?yán)?/li>
- 源碼解讀試?yán)?/li>
- 變換思想圖解
- 變換思想總結(jié)
1 、 簡(jiǎn)單試?yán)?/h3>
Observable.just("url")//
.map(new Function<String, Long>() {
@Override
public Long apply(@NonNull String s) throws Exception {
return 1l;
}
})
.map(new Function<Long, Integer>() {
@Override
public Integer apply(@NonNull Long l) throws Exception {
return 50;
}
})
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull Integer integer) {
Log.i("Info","onNext:"+integer);
}
@Override
public void onError(@NonNull Throwable e) {
Log.i("Info","異常:"+e.getLocalizedMessage());
e.printStackTrace();
}
@Override
public void onComplete() {
}
});
打印:nNext:50
2 、 源碼解讀實(shí)例
Observable.just("url")//
.map(new Function<String, Long>() {
@Override
public Long apply(@NonNull String s) throws Exception {
return 1l;
}
})
.map(new Function<Long, Integer>() {
@Override
public Integer apply(@NonNull Long l) throws Exception {
return 50;
}
})
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull Integer integer) {
Log.i("Info","onNext:"+integer);
}
@Override
public void onError(@NonNull Throwable e) {
Log.i("Info","異常:"+e.getLocalizedMessage());
e.printStackTrace();
}
@Override
public void onComplete() {
}
});
打印:nNext:50
創(chuàng)建 ObservableJust 對(duì)象
public static <T> Observable<T> just(T item) {
ObjectHelper.requireNonNull(item, "The item is null");
//這里其實(shí)是直接返回ObservableJust對(duì)象 .
return RxJavaPlugins.onAssembly(new ObservableJust<T>(item));
}
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
//默認(rèn)情況下 onObservableAssembly 為空 , 需要全局傳入
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
return source;
}
RxJavaPlugins.onAssembly 這里起到了 Hook 作用 , 加多了一層 , 可以用于 Fliter 例如
RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {
@Override
public Observable apply(@NonNull Observable observable) throws Exception {
Log.i("Info","Hook 創(chuàng)建對(duì)象 Observable ");
if(observable instanceof ObservableJust && !((((ObservableJust)(observable)).call()) instanceof String) ){
return null;
}
return observable;
}
});
//再一次運(yùn)行上面的例子就是打印出:Hook 創(chuàng)建對(duì)象 Observable
使得所有屬于全局 ObservableJust 的 value 非 String 類型都無(wú)法正常創(chuàng)建 . 下面進(jìn)入重點(diǎn)變換
變換也是一樣加了一層全局 Hook . 這里直接默認(rèn)返回 new ObservableMap<T, R>(this, mapper)即可
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
ObservableMap 也是 Observable 的一個(gè)子類 , 把 ObservableJust 和 變換函數(shù) Function 傳進(jìn)去創(chuàng)建 ObservableMap , 下一個(gè) ObservableMap 走相同邏輯 , 這樣就形成 ObservableJust(1) ∈ ObservableMap(2)
∈ ObservableMap(3) 的關(guān)系 (∈屬于) 典型的責(zé)任鏈模式 . 上面例子邏輯走到 ···(A).subscribe(new Observer<Integer>()··· (A)處的對(duì)象為ObservableMap(3) , 發(fā)生訂閱 . 繼續(xù)查看源碼
public final void subscribe(Observer<? super T> observer) {
···· 省略N多邏輯
subscribeActual(observer);
subscribeActual是一個(gè)抽象方法 , 查看子類 ObservableMap 實(shí)現(xiàn)該方法
public void subscribeActual(Observer<? super U> t) {
//這里的source為ObservableMap(2)對(duì)象
source.subscribe(new MapObserver<T, U>(t, function));
}
source 調(diào)用 subscribe 跟 ObservableMap(3) 執(zhí)行相同邏輯 , 這里的 source 指的是 ObservableMap(2) 對(duì)象 , 這樣就形成了(有序序列). MainActivity 里的 內(nèi)部類 Observer 被包裝了一層 MapObserver 當(dāng)成參數(shù)往上傳形成了鏈表 MapObserver(3) --> MapObserver(2) --> Observer(1) .
最終回到 ObservableJust 發(fā)生訂閱把數(shù)據(jù)源于流的形式一層一層發(fā)射 , 見
//ObservableJust 的subscribeActual 方法
protected void subscribeActual(Observer<? super T> s) {
Log.i("Info","subscribeActual 發(fā)射 ");
//這里的s 是MapObserver(3)對(duì)象
ScalarDisposable<T> sd = new ScalarDisposable<T>(s, value);
//observer 的 onSubscribe 方法被回調(diào)
s.onSubscribe(sd);
sd.run();
}
public void run() {
if (get() == START && compareAndSet(START, ON_NEXT)) {
//正式發(fā)射數(shù)據(jù)源
observer.onNext(value);
if (get() == ON_NEXT) {
lazySet(ON_COMPLETE);
observer.onComplete();
}
}
}
數(shù)據(jù)源發(fā)射 MapObserver(3).onNext() --(Function變換)--> MapObserver(2).onNext() --(Function 變換)-->Observer(1).onNext() .
3 件缸、 變換思想圖解
說(shuō)明 : subscribe 是發(fā)生在 Observable 里面調(diào)用subscribeActual , 內(nèi)部創(chuàng)建 MapObserver 包裝對(duì)象 Observer , 傳給父級(jí)Observable . source.subscribe(new MapObserver<T, U>(t, function));
4、 變換思想總結(jié).
變換是 RxJava 最核心思想 , RxJava 源碼比較繞 , 但是萬(wàn)變不離其中 , 包括線程調(diào)度 subscribeOn() 和 observeOn() 也是利用變換 .
