0章 RxJava2課程目錄介紹
本篇文章已受權微信公衆號 YYGeeker 獨家發佈轉載請標明出處bash
CSDN學院課程地址微信
- RxJava2從入門到精通-初級篇:edu.csdn.net/course/deta…
- RxJava2從入門到精通-中級篇:edu.csdn.net/course/deta…
- RxJava2從入門到精通-進階篇:edu.csdn.net/course/deta…
- RxJava2從入門到精通-源碼分析篇:edu.csdn.net/course/deta…
10. RxJava源碼分析
RxJava源碼分析最主要的點在於多線程
- RxJava是如何從事件流的發送方發送到事件流的接收方的
- RxJava是如何對操做符進行封裝和操做的
- RxJava是如何隨意切換線程的
在分析的過程當中，部分源碼分析咱們會經過手寫RxJava的部分代碼進行分析，固然也會結合實際RxJava的代碼進行分析。其中，手寫RxJava的緣由是爲了簡化源代碼，讓讀者方便閱讀到主要代碼，更快的看懂RxJava的實現思路。在閱讀源碼以前，咱們須要對RxJava的大致概念進行簡單的梳理app
- 發射器：Emitter，發射數據的對象
- 被觀察者：Observable，被觀察的對象
- 觀察者：Observer，觀察的對象
- 被觀察者被訂閱時：ObservableOnSubscribe，被訂閱時的回調，同時建立出發射器
- 釋放者：Disposable，釋放RxJava的對象
RxJava的分析三步驟composer
- 建立：被觀察者建立的過程
- 訂閱：被觀察者訂閱觀察者的過程
- 發射：發射器發射的過程
RxJava原理圖解框架
- 第一排表示各個對象的建立關係，A->B->C->D
- 第二排表示各個對象的訂閱關係，D->C->B->A
- 第三排表示各個對象的發射關係，A->B->C->D
10.1 RxJava的事件發射原理
知識點：ide
- 理解發射數據的過程
- 理解接收數據的過程
如下是手寫RxJava的代碼模塊化
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(Emitter<String> emitter) {
emitter.onNext("Hello RxJava");
emitter.onError();
emitter.onNext("Hello RxJava");
}
}).subscribe(new Observabler<String>() {
@Override
public void onSubscribe() {
}
@Override
public void onNext(String string) {
System.out.println("onNext=" + string);
}
@Override
public void onError() {
System.out.println("onError");
}
@Override
public void onComplete() {
}
});
複製代碼
輸出結果：在輸出onError後，就不會繼續收到新的事件流，表示事件已經被釋放了函數
onNext=Hello RxJava
onError
複製代碼
一、定義接口工具
發射器
public interface Emitter<T> {
void onNext(T t);
void onError();
}
複製代碼
觀察者
public interface Observer<T> {
void onSubscribe();
void onNext(T t);
void onError();
void onComplete();
}
複製代碼
被觀察者被訂閱時
public interface ObservableOnSubscribe<T> {
void subscribe(Emitter<T> emitter);
}
複製代碼
二、實現被觀察者
被觀察者Observable負責建立、訂閱，發射由發射器負責
- 建立：建立的過程只是將傳遞進來的參數交給新的ObservableCreate進行管理
- 訂閱：訂閱的過程只是實現建立出來的ObservableCreate的subscribeActual方法
public abstract class Observable<T> {
public static <T> ObservableCreate create(ObservableOnSubscribe<T> observableOnSubscribe) {
return new ObservableCreate<T>(observableOnSubscribe);
}
public void subscribe(Observer<T> observer) {
subscribeActual(observer);
}
public abstract void subscribeActual(Observer<T> observer);
}
複製代碼
三、ObservableCreate
ObservableCreate繼承自Observable，因爲Observable.create返回當前ObservableCreate，因此在subscribe的時候，走的是這裏的subscribeActual，subscribeActual中會去建立發射器，並給發射器傳遞進去observer
public class ObservableCreate<T> extends Observable{
private ObservableOnSubscribe source;
public ObservableCreate(ObservableOnSubscribe observableOnSubscribe) {
this.source = observableOnSubscribe;
}
@Override
public void subscribeActual(Observer observer) {
//固定的三步曲分析法（我的建立，基本都是這個步驟）
//一、建立發射器
EmitterCreate<T> emitterCreate = new EmitterCreate<>(observer);
//二、回調observer的onSubscribe
observer.onSubscribe();
//三、回調上一個的subscribe
source.subscribe(emitterCreate);
}
}
複製代碼
四、EmitterCreate
傳遞進來的observer便是咱們最開始訂閱時候new出來的，此時發射數據，就會去調用Observer的onNext方法，這樣數據就從發射器中傳遞到觀察者中了。DisposableHelper在後面會講到，這裏只是用做判斷是否被釋放的一個工具類
public class EmitterCreate<T>
extends AtomicReference<Disposable>
implements Emitter<T>, Disposable {
private Observer<T> observer;
public EmitterCreate(Observer<T> observer) {
this.observer = observer;
}
@Override
public void onNext(T t) {
if (!isDisposed()) {
observer.onNext(t);
}
}
@Override
public void onError() {
if (!isDisposed()) {
try {
observer.onError();
} finally {
dispose();
}
}
}
@Override
public void dispose() {
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
}
複製代碼
如下是RxJava源代碼
一、Observable.create
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");//判空
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));//返回自身
}
複製代碼
RxJavaPlugins.onAssembly只是對傳遞進來的參數作判斷處理，最終仍是返回ObservableCreate，有關RxJavaPlugins的東西最終都是返回自身，RxJavaPlugins後面分析會說到，這裏只須要知道他是返回參數自己便可
二、Observable.subscribe
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");//判空
try {
observer = RxJavaPlugins.onSubscribe(this, observer);//返回自身
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
subscribeActual(observer);//回調ObservableCreate的subscribeActual
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
......
throw npe;
}
}
複製代碼
observable.subscribe和咱們手寫代碼同樣，最終調用的是ObservableCreate的subscribeActual方法
三、ObservableCreate
public final class ObservableCreate<T> extends Observable<T> {
final ObservableOnSubscribe<T> source;
public ObservableCreate(ObservableOnSubscribe<T> source) {
this.source = source;
}
@Override
protected void subscribeActual(Observer<? super T> observer) {
//一、建立發射器
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
//二、回調observer的onSubscribe
observer.onSubscribe(parent);
try {
//三、回調ObservableOnSubscribe的subscribe
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
static final class CreateEmitter<T>
extends AtomicReference<Disposable>
implements ObservableEmitter<T>, Disposable {
private static final long serialVersionUID = -3434801548987643227L;
final Observer<? super T> observer;
CreateEmitter(Observer<? super T> observer) {
this.observer = observer;
}
@Override
public void onNext(T t) {
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
if (!isDisposed()) {
observer.onNext(t);
}
}
@Override
public void onError(Throwable t) {
if (!tryOnError(t)) {
RxJavaPlugins.onError(t);
}
}
@Override
public boolean tryOnError(Throwable t) {
if (t == null) {
t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
}
if (!isDisposed()) {
try {
observer.onError(t);
} finally {
dispose();
}
return true;
}
return false;
}
@Override
public void onComplete() {
if (!isDisposed()) {
try {
observer.onComplete();
} finally {
dispose();
}
}
}
@Override
public void setDisposable(Disposable d) {
DisposableHelper.set(this, d);
}
@Override
public void setCancellable(Cancellable c) {
setDisposable(new CancellableDisposable(c));
}
@Override
public ObservableEmitter<T> serialize() {
return new SerializedEmitter<T>(this);
}
@Override
public void dispose() {
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
}
}
複製代碼
ObservableCreate和咱們手寫代碼同樣，建立發射器，並在發射器中作發射數據等操做
小結
如圖所示
10.2 RxJava的事件釋放原理
知識點：
- 理解釋放事件的原理
有關RxJava的釋放原理是基於Observable能夠返回Disposable對象，只有調用dispose()才能釋放事件，經過上面的例子，咱們知道在發射器裏面有isDisposed()和dispose()操做，在發射完onError事件的狀況下，咱們會將事件釋放，因此在finally會作釋放操做，防止後面的事件再次發射
如下是手寫RxJava的代碼
public class EmitterCreate<T>
extends AtomicReference<Disposable>
implements Emitter<T>, Disposable {
private Observer<T> observer;
public EmitterCreate(Observer<T> observer) {
this.observer = observer;
}
@Override
public void onNext(T t) {
if (!isDisposed()) {
observer.onNext(t);
}
}
@Override
public void onError() {
if (!isDisposed()) {
try {
observer.onError();
} finally {
dispose();
}
}
}
@Override
public void dispose() {
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
}
複製代碼
如下是RxJava源代碼
@Override
public void dispose() {
DisposableHelper.dispose(this);
}
@Override
public boolean isDisposed() {
return DisposableHelper.isDisposed(get());
}
複製代碼
能夠發現事件的釋放都是經過DisposableHelper去觸發的，不論是手寫RxJava仍是源代碼，釋放RxJava都是經過DisposableHelper進行釋放，具體看DisposableHelper。在咱們的演示程序中，咱們經過發射onNext->onError->onNext的過程，去挖掘事件是怎麼被釋放掉的
public enum DisposableHelper implements Disposable {
DISPOSED
;
public static boolean isDisposed(Disposable d) {
return d == DISPOSED;
}
public static boolean dispose(AtomicReference<Disposable> field) {
Disposable current = field.get();//獲取參數的Disposable對象
Disposable d = DISPOSED;//聲明一個已經釋放的Disposable對象
if (current != d) {//若是當前未被釋放
current = field.getAndSet(d);//則將當前的Disposable賦值成已經釋放過的Disposable對象
if (current != d) {//若是當前還未被釋放
if (current != null) {//且不爲空
current.dispose();//則釋放當前Disposable對象
}
return true;
}
}
return false;
}
}
複製代碼
在事件釋放的過程當中，EmitterCreate自己是個AtomicReference<Disposable>，代碼經過get()去獲取Disposable對象，其中代碼會經過雙層判斷去作釋放，防止在多線程的時候出現搶奪的狀況
- onNext：第一次發射數據時，
get()會獲取一個null對象，因此不符合d == DISPOSED - onEroor：這時候會調用
dispose()去比較當前和釋放過的對象，若是不等於，則將當前的對象設置爲釋放過的值 - onNext：第二次發射數據時，
get()會獲取一個已經釋放過的對象，這個時候符合d == DISPOSED
其實這裏的操做如同設置一個Flag，但因爲Disposable是對象的形式，且須要保證原子性，AtomicReference類型是個最佳選擇，能保證對象的原子性
10.3 RxJava的背壓原理
知識點：
- 理解背壓實現的本質
- 理解背壓數據項丟棄的本質
背壓原理有一部分和RxJava事件發射原理類似，其背壓的過程就是在不一樣策略的發射器去處理當前的數據項而已。在分析背壓策略的時候，咱們都知道背壓是須要手動進行請求才能夠將數據發射到觀察者中，因此咱們會調用s.request(Long.MAX_VALUE)讓觀察者能接收到數據。有些人就會有疑問，爲何有些人平時用背壓的時候，不須要去調用request()就能接收到數據，緣由是有些背壓已經在內部默認調用了s.request(Long.MAX_VALUE)，因此這裏是不用多想的，是必定要調用s.request(Long.MAX_VALUE)才能收到數據的。因爲不一樣背壓的策略的原理大同小異，主要以Drop策略去分析背壓的原理
public static void drop(View view) {
Flowable.create(new FlowableOnSubscribe<Integer>() {
@Override
public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
for (int i = 0; i < 1000; i++) {
emitter.onNext(i);
}
}
}, BackpressureStrategy.DROP)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new FlowableSubscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
s.request(Long.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
Log.e("TAG", "onNext=" + integer);
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
@Override
public void onComplete() {
}
});
}
複製代碼
如下是RxJava源代碼
一、Flowable.create
public static <T> Flowable<T> create(FlowableOnSubscribe<T> source, BackpressureStrategy mode) {
ObjectHelper.requireNonNull(source, "source is null");//判空
ObjectHelper.requireNonNull(mode, "mode is null");//判空
return RxJavaPlugins.onAssembly(new FlowableCreate<T>(source, mode));//返回自身
}
複製代碼
Flowable.create跟咱們前面是同樣的，最後仍是會交給新對象FlowableCreate去處理
二、Flowable.subscribe
public final void subscribe(FlowableSubscriber<? super T> s) {
ObjectHelper.requireNonNull(s, "s is null");
try {
Subscriber<? super T> z = RxJavaPlugins.onSubscribe(this, s);
ObjectHelper.requireNonNull(z, "Plugin returned null Subscriber");
subscribeActual(z);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
......
throw npe;
}
}
複製代碼
Flowable.subscribe跟咱們前面是同樣的，最終調用的是FlowableCreate的subscribeActual方法
三、FlowableCreate.subscribeActual
public final class FlowableCreate<T> extends Flowable<T> {
final FlowableOnSubscribe<T> source;
final BackpressureStrategy backpressure;
public FlowableCreate(FlowableOnSubscribe<T> source, BackpressureStrategy backpressure) {
this.source = source;
this.backpressure = backpressure;
}
@Override
public void subscribeActual(Subscriber<? super T> t) {
//使用三步曲分析法
BaseEmitter<T> emitter;
switch (backpressure) {
case MISSING: {
emitter = new MissingEmitter<T>(t);
break;
}
case ERROR: {
emitter = new ErrorAsyncEmitter<T>(t);
break;
}
case DROP: {
emitter = new DropAsyncEmitter<T>(t);
break;
}
case LATEST: {
emitter = new LatestAsyncEmitter<T>(t);
break;
}
default: {
emitter = new BufferAsyncEmitter<T>(t, bufferSize());
break;
}
}
t.onSubscribe(emitter);
try {
source.subscribe(emitter);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
emitter.onError(ex);
}
}
}
複製代碼
subscribeActual會根據不一樣的策略生成不一樣的發射器，具體的全部策略邏輯都在發射器中體現的
四、DropAsyncEmitter
static final class DropAsyncEmitter<T> extends NoOverflowBaseAsyncEmitter<T> {
private static final long serialVersionUID = 8360058422307496563L;
DropAsyncEmitter(Subscriber<? super T> actual) {
super(actual);
}
@Override
void onOverflow() {
// nothing to do
}
}
複製代碼
DropAsyncEmitter其實沒作什麼事情，主要都在其父類中實現了，onOverflow的回調錶示事件流溢出的時候的處理，很明顯Drop策略就把溢出的數據項直接不作處理，意思就是拋棄掉這個數據項了
static final class ErrorAsyncEmitter<T> extends NoOverflowBaseAsyncEmitter<T> {
private static final long serialVersionUID = 338953216916120960L;
ErrorAsyncEmitter(Subscriber<? super T> actual) {
super(actual);
}
@Override
void onOverflow() {
onError(new MissingBackpressureException("create: could not emit value due to lack of requests"));
}
}
複製代碼
再看看Error策略，溢出以後就會拋出溢出的異常，其餘策略也相似分析，具體父類是如何處理溢出函數的呢
五、NoOverflowBaseAsyncEmitter
abstract static class NoOverflowBaseAsyncEmitter<T> extends BaseEmitter<T> {
private static final long serialVersionUID = 4127754106204442833L;
NoOverflowBaseAsyncEmitter(Subscriber<? super T> actual) {
super(actual);
}
@Override
public final void onNext(T t) {
if (isCancelled()) {
return;
}
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
//這裏暫時將get()函數看成是相似於List這種的容器，存儲的是當前須要處理的數據項
if (get() != 0) { //從數據項容器中取值，若是當前有數據項須要處理
actual.onNext(t); //發射數據項
BackpressureHelper.produced(this, 1); //對當前存在須要處理的數據項進行-1操做
} else {
onOverflow(); //從數據項容器中取值，若是當前沒有數據項須要處理，則回調溢出函數
}
}
abstract void onOverflow();
}
複製代碼
NoOverflowBaseAsyncEmitter在發射數據項的時候，會去BaseEmitter中的數據項容器去取出數據項，若是存在則處理，不存在則表示溢出，回調溢出函數，那麼具體的數據項容器時候怎麼存儲須要處理的數據項的呢
六、BaseEmitter
abstract static class BaseEmitter<T>
extends AtomicLong
implements FlowableEmitter<T>, Subscription {
private static final long serialVersionUID = 7326289992464377023L;
final Subscriber<? super T> actual;
final SequentialDisposable serial;
BaseEmitter(Subscriber<? super T> actual) {
this.actual = actual;
this.serial = new SequentialDisposable();
}
@Override
public void onComplete() {
complete();
}
protected void complete() {
if (isCancelled()) {
return;
}
try {
actual.onComplete();
} finally {
serial.dispose();
}
}
@Override
public final void onError(Throwable e) {
if (!tryOnError(e)) {
RxJavaPlugins.onError(e);
}
}
@Override
public final void request(long n) {
if (SubscriptionHelper.validate(n)) {
BackpressureHelper.add(this, n);
}
}
}
複製代碼
BaseEmitter就是一個AtomicLong，若是沒學過AtomicLong的話，能夠簡單理解爲一個計數器，get()就是獲取當前的Long值，只要不等於0就表示有值。主要仍是在request()，request()表示此時須要處理的數據項。結合上面NoOverflowBaseAsyncEmitter的中的BackpressureHelper.produced(this, 1)和當前BaseEmitter中的BackpressureHelper.add(this, n)，可得數據項的容器徹底都是由BackpressureHelper去控制，咱們只須要對BackpressureHelper的存儲和獲取作分析，就能夠知道當前是否有數據項須要處理
七、BackpressureHelper
public final class BackpressureHelper {
public static long add(AtomicLong requested, long n) {
for (;;) {
long r = requested.get(); //獲取當前數據項
if (r == Long.MAX_VALUE) {
return Long.MAX_VALUE;
}
long u = addCap(r, n);//當前數據項 + 新增的數據項
if (requested.compareAndSet(r, u)) { //設置最新的數據項
return r;
}
}
}
public static long addCap(long a, long b) {
long u = a + b;
if (u < 0L) {
return Long.MAX_VALUE;
}
return u;
}
public static long produced(AtomicLong requested, long n) {
for (;;) {
long current = requested.get(); //獲取當前數據項
if (current == Long.MAX_VALUE) {
return Long.MAX_VALUE;
}
long update = current - n; //當前數據項 - 須要發射的數據項（從源碼上，n爲1）
if (update < 0L) { //不能爲負數
RxJavaPlugins.onError(new IllegalStateException("More produced than requested: " + update));
update = 0L;
}
if (requested.compareAndSet(current, update)) { //設置最新的數據項
return update;
}
}
}
}
複製代碼
BackpressureHelper就是利用AtomicLong的原子性就行簡單的計數器操做而已，並無什麼複雜的操做。至此，咱們就知道背壓的原理原來就是利用AtomicLong計數器和生產消費的模式去決定是否發射當前的數據項而已
10.4 RxJava的經常使用操做符原理
知識點：
- 理解map操做符的原理
RxJava經常使用操做符的表明就是map，分析map源碼後，其餘的操做符的思想是同樣的，只不過是實現邏輯不一致而已。下面咱們經過分析map的主要流程去分析map是如何轉換字符串的，從上面咱們知道Observable的建立、訂閱、發射的過程，此次對於重複的內容就再也不繼續分析，主要是分析中間map是如何回調apply()去將數據項轉換成字符串的
public void map() {
//建立被觀察者
Observable
.create(new ObservableOnSubscribe<String>() {
@Override
//默認在主線程裏執行該方法
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
e.onNext("俊俊俊很帥");
e.onNext("你值得擁有");
e.onNext("取消關注");
e.onNext("但仍是要保持微笑");
e.onComplete();
}
})
.map(new Function<String, String>() {
@Override
public String apply(String s) throws Exception {
return "Hello";
}
})
//建立觀察者並訂閱
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
if (!d.isDisposed()) {
d.dispose();
}
}
@Override
public void onNext(String s) {
System.out.println("onNext=" + s);
}
@Override
public void onError(Throwable e) {
System.out.println("onNext=" + e.getMessage());
}
@Override
public void onComplete() {
}
});
}
複製代碼
如下是RxJava源代碼
一、Observable.map
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
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));
}
複製代碼
從create到map的過程當中，create的時候，當前的Observable已經被轉換成ObservableCreate，再次map的時候， 當前的Observable已經被轉換成ObservableMap，並且在ObservableMap中傳遞的參數包含this，因此當前ObservableMap中是嵌套着ObservableCreate
二、Observable.subscribe
因爲當前的Observable是ObservableMap，因此Observable.subscribe會回調ObservableMap中的subscribeActual
public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends U> function;
public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
super(source);
this.function = function;
}
@Override
public void subscribeActual(Observer<? super U> t) {
source.subscribe(new MapObserver<T, U>(t, function));//source是傳遞進來的ObservableCreate
}
}
複製代碼
ObservableMap中的subscribeActual，會去調用ObservableCreate的subscribe方法，最後仍是會去回調 ObservableCreate的subscribeActual，不過這裏在回調的過程當中增長了一個參數MapObserver，這個參數只有在ObservableCreate發射器發射的時候纔會被調用
三、ObservableCreate.subscribeActual
@Override
protected void subscribeActual(Observer<? super T> observer) {
CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);
try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}
}
複製代碼
在ObservableCreate.subscribeActual中，會接收一個Observer的參數，這個時候的Observer的參數是從ObservableMap中傳遞過來的MapObserver，當CreateEmitter發射onNext的時候，就會在當前的MapObserver對象onNext進行處理
四、MapObserver.onNext
static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;
MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
super(actual);
this.mapper = mapper;
}
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != NONE) {
actual.onNext(null);
return;
}
U v;
try {
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
actual.onNext(v);
}
......
}
複製代碼
onNext主要作了兩個事情，一個是mapper.apply(t)，這個就是map操做符所實現的方法，這裏就將原來的值轉換成新的值，一個是actual.onNext(v)，將轉換出來的新值v繼續onNext出去，這裏的actual就是在構造函數中傳遞進來的ObservableCreate，這裏就已經將數據項通過map的操做符後繼續執行後面正常的發射流程
小結
如圖所示
10.5 RxJava的線程切換原理
知識點：
- 理解在工做線程上爲何能執行耗時操做
- 理解在UI線程爲何能執行更新UI的操做
沿用上面的例子，在線程切換的過程當中，無非就是至關於不一樣的操做符繼續操做數據項而已，根本的實現思路和map等操做符是同樣的，也是經過嵌套Observable的過程來執行的，只不過是線程切換的操做符內部實現的邏輯有區別而已。經過咱們以往的思路去想，這兩個知識點無非就是啓動線程池去執行耗時任務，而UI線程則是交給Handler去處理，RxJava線程切換的原理就是這樣的
Observable
.create(new ObservableOnSubscribe<String>() {
@Override
//默認在主線程裏執行該方法
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
e.onNext("俊俊俊很帥");
e.onNext("你值得擁有");
e.onNext("取消關注");
e.onNext("但仍是要保持微笑");
e.onComplete();
}
})
.map(new Function<String, String>() {
@Override
public String apply(String s) throws Exception {
return "Hello";
}
})
//將被觀察者切換到子線程
.subscribeOn(Schedulers.io())
//將觀察者切換到主線程 須要在Android環境下運行
.observeOn(AndroidSchedulers.mainThread())
//建立觀察者並訂閱
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
if (!d.isDisposed()) {
d.dispose();
}
}
@Override
public void onNext(String s) {
System.out.println("onNext=" + s);
}
@Override
public void onError(Throwable e) {
System.out.println("onNext=" + e.getMessage());
}
@Override
public void onComplete() {
}
});
複製代碼
基礎概念：
- Schedulers：調度器的管理者。管理着多種不一樣種類的Scheduler
- Scheduler：調度器。負責線程Worker的建立
createWorker()，調度Worker的執行schedule() - Worker：抽象的工做線程。被線程調度器管理，負責線程的建立和執行
在源碼中，咱們須要先熟悉這三者之間的關係究竟是如何運做的
如下是RxJava源代碼
一、observeOn()
@CheckReturnValue
@SchedulerSupport("custom")
public final Observable<T> observeOn(Scheduler scheduler) {
return this.observeOn(scheduler, false, bufferSize());
}
@CheckReturnValue
@SchedulerSupport("custom")
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(new ObservableObserveOn(this, scheduler, delayError, bufferSize));
}
複製代碼
當前的Observable已經被轉換成ObservableObserveOn
public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
final boolean delayError;
final int bufferSize;
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler;
this.delayError = delayError;
this.bufferSize = bufferSize;
}
protected void subscribeActual(Observer<? super T> observer) {
if (this.scheduler instanceof TrampolineScheduler) {
this.source.subscribe(observer);
} else {
//一、建立工做線程
Worker w = this.scheduler.createWorker();
//二、訂閱以後，在發射的過程當中
this.source.subscribe(new ObservableObserveOn.ObserveOnObserver(observer, w, this.delayError, this.bufferSize));
}
}
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable {
private static final long serialVersionUID = 6576896619930983584L;
final Observer<? super T> actual;
final Worker worker;
final boolean delayError;
final int bufferSize;
SimpleQueue<T> queue;
public void onNext(T t) {
if (!this.done) {
if (this.sourceMode != 2) {
this.queue.offer(t);
}
//三、在OnNext中執行
this.schedule();
}
}
void schedule() {
if (this.getAndIncrement() == 0) {
//四、執行工做線程
this.worker.schedule(this);
}
}
}
}
複製代碼
其三者的關係簡單的說就是在每次訂閱的時候，都會去建立出對應的工做線程，這個工做線程取決於你傳遞的參數是哪一個Worker，在發射器發射的過程當中，這個工做線程總會去執行它的回調schedule，其實大部分的操做就是在schedule裏面執行線程。搞懂了三者的關係以後，分析線程切換就簡單多了，就至關於工廠同樣，給個具體的任務給到具體的工人去執行，很像工廠的流水線，咱們已經肯定下來了流水線的流程了，這個時候咱們就須要去關心參數具體是什麼東西了。在閱讀subscribeOn、observeOn前，咱們先看看這兩個方法中的參數都是什麼
一、Schedulers.io()
public final class Schedulers {
@NonNull
static final Scheduler IO;
static {
SINGLE = RxJavaPlugins.initSingleScheduler(new SingleTask());
COMPUTATION = RxJavaPlugins.initComputationScheduler(new ComputationTask());
//一、在初始化的時候就構建出了IOTask，initIoScheduler會去執行IOTask的call方法
IO = RxJavaPlugins.initIoScheduler(new IOTask());
TRAMPOLINE = TrampolineScheduler.instance();
NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask());
}
static final class IOTask implements Callable<Scheduler> {
@Override
public Scheduler call() throws Exception {
//二、IOTask的call方法會去獲取IoHolder的值
return IoHolder.DEFAULT;
}
}
static final class IoHolder {
//三、建立IoScheduler
static final Scheduler DEFAULT = new IoScheduler();
}
public static Scheduler io() {
//Schedulers.io()：它會去獲取前面3步建立出來的IoScheduler對象
return RxJavaPlugins.onIoScheduler(IO); //返回IO自身
}
}
複製代碼
其正在實如今IoScheduler，其表示管理Io線程的管理者
public final class IoScheduler extends Scheduler {
final AtomicReference<CachedWorkerPool> pool;
static final CachedWorkerPool NONE;
static {
......
//一、建立CachedWorkerPool
NONE = new CachedWorkerPool(0, null, WORKER_THREAD_FACTORY);
}
static final class CachedWorkerPool implements Runnable {
......
private final ScheduledExecutorService evictorService;
private final Future<?> evictorTask;
CachedWorkerPool(long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory) {
......
Future<?> task = null;
if (unit != null) {
evictor = Executors.newScheduledThreadPool(1, EVICTOR_THREAD_FACTORY);
task = evictor.scheduleWithFixedDelay(this, this.keepAliveTime, this.keepAliveTime, TimeUnit.NANOSECONDS);
}
evictorService = evictor;
evictorTask = task;
}
}
@NonNull
@Override
public Worker createWorker() {
//二、建立具體的線程
return new EventLoopWorker(pool.get());
}
static final class EventLoopWorker extends Scheduler.Worker {
private final CompositeDisposable tasks;
private final CachedWorkerPool pool;
private final ThreadWorker threadWorker;
final AtomicBoolean once = new AtomicBoolean();
EventLoopWorker(CachedWorkerPool pool) {
this.pool = pool;
this.tasks = new CompositeDisposable();
this.threadWorker = pool.get();
}
@NonNull
@Override
public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
if (tasks.isDisposed()) {
// don't schedule, we are unsubscribed return EmptyDisposable.INSTANCE; } //三、最終會去調用ThreadWorker的scheduleActual return threadWorker.scheduleActual(action, delayTime, unit, tasks); } } //四、因爲ThreadWorker沒有scheduleActual，在父類中找NewThreadWorker static final class ThreadWorker extends NewThreadWorker { private long expirationTime; ThreadWorker(ThreadFactory threadFactory) { super(threadFactory); this.expirationTime = 0L; } public long getExpirationTime() { return expirationTime; } public void setExpirationTime(long expirationTime) { this.expirationTime = expirationTime; } } } 複製代碼
NewThreadWorker，最終仍是調用executor.submit()或executor.schedule()
@NonNull
public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);
if (parent != null) {
if (!parent.add(sr)) {
return sr;
}
}
Future<?> f;
try {
if (delayTime <= 0) {
f = executor.submit((Callable<Object>)sr);
} else {
f = executor.schedule((Callable<Object>)sr, delayTime, unit);
}
sr.setFuture(f);
} catch (RejectedExecutionException ex) {
if (parent != null) {
parent.remove(sr);
}
RxJavaPlugins.onError(ex);
}
return sr;
}
複製代碼
二、AndroidSchedulers.mainThread()
public final class AndroidSchedulers {
private static final class MainHolder {
static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));
}
private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
new Callable<Scheduler>() {
@Override public Scheduler call() throws Exception {
return MainHolder.DEFAULT;
}
});
/** A {@link Scheduler} which executes actions on the Android main thread. */
public static Scheduler mainThread() {
return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
}
}
複製代碼
返回一個HandlerScheduler，建立單例模式的主線程Handler
final class HandlerScheduler extends Scheduler {
private final Handler handler;
HandlerScheduler(Handler handler) {
this.handler = handler;
}
public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
if (run == null) {
throw new NullPointerException("run == null");
} else if (unit == null) {
throw new NullPointerException("unit == null");
} else {
run = RxJavaPlugins.onSchedule(run);
HandlerScheduler.ScheduledRunnable scheduled = new HandlerScheduler.ScheduledRunnable(this.handler, run);
this.handler.postDelayed(scheduled, unit.toMillis(delay));
return scheduled;
}
}
public Worker createWorker() {
//建立具體工做線程
return new HandlerScheduler.HandlerWorker(this.handler);
}
......
}
複製代碼
就好像咱們上面分析的三者關係同樣，Schedule最終仍是會管理着具體的工做線程
private static final class HandlerWorker extends Worker {
private final Handler handler;
private volatile boolean disposed;
HandlerWorker(Handler handler) {
this.handler = handler;
}
@Override
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
if (run == null) throw new NullPointerException("run == null");
if (unit == null) throw new NullPointerException("unit == null");
if (disposed) {
return Disposables.disposed();
}
run = RxJavaPlugins.onSchedule(run);
//包裝新的Runnable交給Handler
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this; // Used as token for batch disposal of this worker's runnables. handler.sendMessageDelayed(message, unit.toMillis(delay)); // Re-check disposed state for removing in case we were racing a call to dispose(). if (disposed) { handler.removeCallbacks(scheduled); return Disposables.disposed(); } return scheduled; } @Override public void dispose() { disposed = true; handler.removeCallbacksAndMessages(this /* token */); } @Override public boolean isDisposed() { return disposed; } } 複製代碼
三、subscribeOn()
理解完參數後，回到咱們的分析重點
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
複製代碼
subscribeOn就如同普通操做符同樣，包裝一層ObservableSubscribeOn，在subscribe的時候真正走的仍是subscribeActual
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
@Override
public void subscribeActual(final Observer<? super T> s) {
//使用三步曲分析法
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
s.onSubscribe(parent);
//三、將第三步的內容放到線程中去執行
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;
SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}
@Override
public void run() {
//三、回調ObservableOnSubscribe的subscribe
source.subscribe(parent);
}
}
}
複製代碼
scheduler.scheduleDirect中會去執行Scheduler裏的方法，這裏的scheduler就是IoScheduler
@NonNull
public Disposable scheduleDirect(@NonNull Runnable run) {
return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
}
@NonNull
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
final Worker w = createWorker();
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
DisposeTask task = new DisposeTask(decoratedRun, w);
w.schedule(task, delay, unit);
return task;
}
複製代碼
回調IoScheduler的createWorker()並執行w.schedule()
小結
如圖所示
10.6 RxJava的自定義Operator原理
知識點：
- 自定義Operator是如何實現的
在講解以前，讓咱們先回味下自定義Operator
public class CustomOperator implements ObservableOperator<String, List<String>> {
@Override
public Observer<? super List<String>> apply(final Observer<? super String> observer) throws Exception {
return new Observer<List<String>>() {
@Override
public void onSubscribe(Disposable d) {
observer.onSubscribe(d);
}
@Override
public void onNext(List<String> strings) {
observer.onNext(strings.toString());
}
@Override
public void onError(Throwable e) {
observer.onError(e);
}
@Override
public void onComplete() {
observer.onComplete();
}
};
}
}
複製代碼
Observable.create(new ObservableOnSubscribe<List<String>>() {
@Override
public void subscribe(@NonNull ObservableEmitter<List<String>> e) throws Exception {
}
}).lift(new CustomOperator())
複製代碼
自定義Operator如同普通的操做符原理差很少，用的是lift的操做符，只不過在lift裏面將邏輯的執行回調到自定義的Operator的apply()
如下是RxJava源代碼
一、Observable.lift
public final <R> Observable<R> lift(ObservableOperator<? extends R, ? super T> lifter) {
ObjectHelper.requireNonNull(lifter, "onLift is null");
return RxJavaPlugins.onAssembly(new ObservableLift<R, T>(this, lifter));
}
複製代碼
二、ObservableLift.subscribeActual
public final class ObservableLift<R, T> extends AbstractObservableWithUpstream<T, R> {
/** The actual operator. */
final ObservableOperator<? extends R, ? super T> operator;
public ObservableLift(ObservableSource<T> source, ObservableOperator<? extends R, ? super T> operator) {
super(source);
this.operator = operator;
}
@Override
public void subscribeActual(Observer<? super R> s) {
Observer<? super T> observer;
try {
observer = ObjectHelper.requireNonNull(operator.apply(s), "Operator " + operator + " returned a null Observer");
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
// can't call onError because no way to know if a Disposable has been set or not // can't call onSubscribe because the call might have set a Disposable already
RxJavaPlugins.onError(e);
NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
npe.initCause(e);
throw npe;
}
source.subscribe(observer);
}
}
複製代碼
能夠看到代碼很是快的就將傳遞進來的參數operator執行apply()
10.7 RxJava的自定義Transformer原理
知識點：
- 自定義Transformer是如何實現的
在講解以前，讓咱們先回味下自定義Transformer
public class NetWorkTransformer implements ObservableTransformer {
@Override
public ObservableSource apply(Observable upstream) {
return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
}
}
複製代碼
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
}
}).compose(new CustomTransformer())
複製代碼
自定義Transformer如同普通的操做符原理差很少，用的是compose的操做符，只不過在compose裏面將邏輯的執行回調到自定義的Transformer的apply()
如下是RxJava源代碼
一、Observable.compose
public final <R> Observable<R> compose(ObservableTransformer<? super T, ? extends R> composer) {
return wrap(((ObservableTransformer<T, R>) ObjectHelper.requireNonNull(composer, "composer is null")).apply(this));
}
複製代碼
能夠看到代碼很是快的就將傳遞進來的參數composer執行apply()，這裏的wrap()只是將代碼裹了一層，若是你想簡單的理解的話，能夠理解爲做者的強迫症犯了，只是爲了讓全部代碼看起來都比較規範，否則這裏實在和其餘操做符的實現不同，咱們能夠追進去wrap()
public static <T> Observable<T> wrap(ObservableSource<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
if (source instanceof Observable) {
return RxJavaPlugins.onAssembly((Observable<T>)source);
}
return RxJavaPlugins.onAssembly(new ObservableFromUnsafeSource<T>(source));
}
複製代碼
wrap()實際上是對composer操做符作了Hook，由於全部操做符都會被RxJava去Hook住，這裏會在下面講到自定義Plugin原理的時候就明白了
10.8 RxJava的自定義Plugin原理
知識點：
- 自定義Plugin是如何實現AOP的
在講解以前，讓咱們先回味下自定義Plugin
public class CustomObservableAssembly implements Function<Observable, Observable> {
@Override
public Observable apply(Observable observable) throws Exception {
System.out.println("CustomObservableAssembly observable.toString:" + observable.toString());
return observable;
}
}
複製代碼
RxJavaPlugins.setOnObservableAssembly(new CustomObservableAssembly());
複製代碼
在自定義Plugin中，相似於Android的術語Hook，但在這裏並非真正的Hook，而是做者在寫RxJava的時候去限定一套規範，讓後面的全部操做符或其餘操做等，均可以實現Hook的原理
如下是RxJava源代碼
一、RxJavaPlugins.setOnObservableAssembly
public static void setOnObservableAssembly(@Nullable Function<? super Observable, ? extends Observable> onObservableAssembly) {
if (lockdown) {
throw new IllegalStateException("Plugins can't be changed anymore");
}
RxJavaPlugins.onObservableAssembly = onObservableAssembly;
}
複製代碼
RxJavaPlugins.setOnObservableAssembly只是對成員變量設置了自定義的值，這個時候onObservableAssembly就有了值，默認是爲null的。設置完值就表示已經Hook成功了，當操做符執行的時候，是如何回調咱們Hook的函數的
二、Observable.create
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
複製代碼
create至關於一個操做符，在每一個操做符的裏面都會去執行一段RxJavaPlugins.onAssembly，這裏就是RxJava規定的規範，一開始咱們只是說返回自身，可是有了Hook以後，就會回調Hook函數，返回已經通過二次加工的自身
三、RxJavaPlugins.onAssembly
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
return apply(f, source);
}
return source;
}
複製代碼
因爲咱們已經設置了新值，這裏的onObservableAssembly就不爲null，不爲null則執行apply()，apply()就是咱們Hook傳進去參數的回調方法
10.9 美團WhiteBoard
美團的WhiteBoard實際上是取自美團的開源框架Shield——開源的移動端頁面模塊化開發框架中的代碼，其主要做用是應用RxJava的Subject搭起組件間通信的橋樑。實質上在WhiteBoard中，是將全部的組件的數據和Subject通信的橋樑保存起來，經過key做爲組件的惟一標誌。不過比較惋惜的是WhiteBoard使用的是RxJava1，不過關係不大，只要讀懂裏面的源碼便可
一、WhiteBoard的初始化
初始化放在Activity/Fragment界面中，至關於通信的橋樑，每一個界面中僅有一個WhiteBoard的實例，並由全部組件共用
public abstract class ShieldFragment extends Fragment implements AgentCellBridgeInterface, DriverInterface {
static final String TAG = ShieldFragment.class.getSimpleName();
......
protected WhiteBoard whiteBoard;
public ShieldFragment() {
this.whiteBoard = new WhiteBoard();//初始化
}
@Override
public void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
......
whiteBoard.onCreate(savedInstanceState);//對應生命週期
}
@Override
public void onDestroy() {
super.onDestroy();
......
whiteBoard.onDestory();//對應生命週期
}
@Override
public void onSaveInstanceState(Bundle outState) {
super.onSaveInstanceState(outState);
......
whiteBoard.onSaveInstanceState(outState);//對應生命週期
}
@Override
public WhiteBoard getWhiteBoard() {
return whiteBoard;//獲取實例
}
}
複製代碼
二、WhiteBoard監聽通知
組件只監聽某個key的事件，有通知的時候就能收到
public class MixCellAgent extends LightAgent {
private MixCell mixCell;
private Subscription loadingSubscription;
private Subscription emptySubscription;
public MixCellAgent(Fragment fragment, DriverInterface bridge, PageContainerInterface pageContainer) {
super(fragment, bridge, pageContainer);
}
@Override
public void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
mixCell = new MixCell(getContext(), this);
loadingSubscription = getWhiteBoard().getObservable(MixLoadingAgent.KEY_LOADING).filter(new Func1() {
@Override
public Object call(Object o) {
return o instanceof Boolean && ((Boolean) o);
}
}).subscribe(new Action1() {
@Override
public void call(Object o) {
loading();
}
});
emptySubscription = getWhiteBoard().getObservable(MixLoadingAgent.KEY_EMPTY).filter(new Func1<Object, Boolean>() {
@Override
public Boolean call(Object o) {
return o instanceof Boolean && ((Boolean) o);
}
}).subscribe(new Action1() {
@Override
public void call(Object o) {
mixCell.onEmpty();
}
});
......
}
@Override
public void onDestroy() {
if (loadingSubscription != null) {
loadingSubscription.unsubscribe();
loadingSubscription = null;
}
if (emptySubscription != null) {
emptySubscription.unsubscribe();
}
......
}
}
複製代碼
三、WhiteBoard的發送通知
WhiteBoard發送通知就是調用WhiteBoard提供的全部put方法，具體是如何收到消息的，還須要經過WhiteBoard的源碼看下
public class MixLoadingAgent extends LightAgent implements MixLoadingCell.MixLoadingListener {
public static final String KEY_LOADING = "loading";
public static final String KEY_EMPTY = "empty";
public static final String KEY_FAILED = "failed";
public static final String KEY_MORE = "more";
public static final String KEY_DONE = "done";
private MixLoadingCell mixLoadingCell;
public MixLoadingAgent(Fragment fragment, DriverInterface bridge, PageContainerInterface pageContainer) {
super(fragment, bridge, pageContainer);
mixLoadingCell = new MixLoadingCell(getContext());
mixLoadingCell.setOnMixLoadingListener(this);
}
@Override
public SectionCellInterface getSectionCellInterface() {
return mixLoadingCell;
}
@Override
public void onLoading() {
getWhiteBoard().putBoolean(KEY_LOADING, true);
}
@Override
public void onEmpty() {
getWhiteBoard().putBoolean(KEY_EMPTY, true);
}
@Override
public void onFailed() {
getWhiteBoard().putBoolean(KEY_FAILED, true);
}
@Override
public void onMore() {
getWhiteBoard().putBoolean(KEY_MORE, true);
}
@Override
public void onDone() {
getWhiteBoard().putBoolean(KEY_DONE, true);
}
}
複製代碼
四、WhiteBoard的原理
最後只須要獲取實例後發送通知便可，getWhiteBoard().putBoolean(key)。WhiteBoard原理是隻要仍是Subject的橋樑的做用
public class WhiteBoard {
public static final String WHITE_BOARD_DATA_KEY = "White_Board_Data";
protected Bundle mData;//保存全部組件的數據
protected HashMap<String, Subject> subjectMap;//保存全部組件的通信橋樑
public WhiteBoard() {
this(null);
}
public WhiteBoard(Bundle data) {
mData = data;
if (mData == null) {
mData = new Bundle();//初始化
}
subjectMap = new HashMap<>();//初始化
}
public void onCreate(Bundle savedInstanceState) {
if (savedInstanceState != null) {
mData = savedInstanceState.getBundle(WHITE_BOARD_DATA_KEY);
}
if (mData == null) {
mData = new Bundle();
}
}
public void onSaveInstanceState(Bundle outState) {
if (outState != null) {
// here we must save a new copy of the mData into the outState
outState.putBundle(WHITE_BOARD_DATA_KEY, new Bundle(mData));
}
}
public void onDestory() {
subjectMap.clear();
mData.clear();
}
//經過key獲取某組件的橋樑
public Observable getObservable(final String key) {
Subject res = null;
if (subjectMap.containsKey(key)) {
res = subjectMap.get(key);
} else {
res = PublishSubject.create();
subjectMap.put(key, res);
}
if (getData(key) != null) {
return res.startWith(getData(key));//帶上已經存儲過的數據
} else {
return res;
}
}
//經過key通知某組件
protected void notifyDataChanged(String key) {
if (subjectMap.containsKey(key)) {
subjectMap.get(key).onNext(mData.get(key));
}
}
//移除組件中的數據
public void removeData(String key) {
mData.remove(key);
notifyDataChanged(key);
}
//每次put值的時候，就會去通知對應的組件
public void putBoolean(@Nullable String key, boolean value) {
mData.putBoolean(key, value);
notifyDataChanged(key);
}
public void putInt(@Nullable String key, int value) {
mData.putInt(key, value);
notifyDataChanged(key);
}
public void putString(@Nullable String key, @Nullable String value) {
mData.putString(key, value);
notifyDataChanged(key);
}
......
public double getDouble(String key) {
return mData.getDouble(key);
}
public String getString(String key, String defaultValue) {
return mData.getString(key, defaultValue);
}
......
}
複製代碼
相關文章
- 1. 第一章 課程介紹
- 2. 0課程介紹（Week1，3月3日）
- 3. Maven工程目錄介紹
- 4. Keras 入門課0 -- 目錄
- 5. 課程介紹
- 6. Angular項目目錄介紹
- 7. 小程序開發第二課——工程文件目錄介紹
- 8. laravel目錄介紹
- 9. CentOs目錄介紹
- 10. CentOS 目錄介紹
- 更多相關文章...
- • Spring目錄結構和基礎JAR包介紹 - Spring教程
- • 網站主機 介紹 - 網站主機教程
- • Java Agent入門實戰（一）-Instrumentation介紹與使用
- • Java 8 Stream 教程
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 第一章 課程介紹
- 2. 0課程介紹（Week1，3月3日）
- 3. Maven工程目錄介紹
- 4. Keras 入門課0 -- 目錄
- 5. 課程介紹
- 6. Angular項目目錄介紹
- 7. 小程序開發第二課——工程文件目錄介紹
- 8. laravel目錄介紹
- 9. CentOs目錄介紹
- 10. CentOS 目錄介紹