目錄
【RxJava】- 創(chuàng)建操作符源碼分析
【RxJava】- 變換操作符源碼分析
【RxJava】- 過(guò)濾操作符源碼分析
【RxJava】- 連接操作符源碼分析
CombineLatest
作用于最近發(fā)射的數(shù)據(jù)項(xiàng)：如果Observable1發(fā)射了A并且Observable2發(fā)射了B和C，combineLatest()將會(huì)分組處理AB和AC据块，處理方式是使用一個(gè)函數(shù)結(jié)合每個(gè)Observable發(fā)射的最近數(shù)據(jù)項(xiàng)，并且基于這個(gè)函數(shù)的結(jié)果發(fā)射數(shù)據(jù)闪朱。
RxJava將這個(gè)操作符實(shí)現(xiàn)為combineLatest，它接受二到九個(gè)Observable作為參數(shù)谆趾，或者單個(gè)Observables列表作為參數(shù)庵芭。它默認(rèn)不在任何特定的調(diào)度器上執(zhí)行。
由
ObservableCombineLatest
實(shí)現(xiàn)
看一下ObservableCombineLatest$LatestCoordinator中的innerNext闹蒜，combineLatest操作符實(shí)現(xiàn)邏輯在這個(gè)內(nèi)部類中。
void innerNext(int index, T item) {
boolean shouldDrain = false;
synchronized (this) {
Object[] latest = this.latest;
if (latest == null) {
return;
}
Object o = latest[index];
int a = active;
if (o == null) {
active = ++a;
}
latest[index] = item;
if (a == latest.length) {
queue.offer(latest.clone());
shouldDrain = true;
}
}
if (shouldDrain) {
drain();
this.latest數(shù)組是在創(chuàng)建LatestCoordinator對(duì)象時(shí)抑淫，根據(jù)combineLatest方法Observable參數(shù)個(gè)數(shù)為數(shù)組長(zhǎng)度創(chuàng)建的數(shù)組绷落。
每一個(gè)Observable（發(fā)射數(shù)據(jù)的Observable，即被觀察者）都有自己的index始苇，當(dāng)發(fā)射數(shù)據(jù)時(shí)砌烁，先從緩存數(shù)組取該索引對(duì)應(yīng)的值，如果存在者更新，如果不存在者計(jì)數(shù)active變量加1函喉，如果計(jì)數(shù)變量值等于數(shù)組長(zhǎng)度避归，表示每一個(gè)Observable都發(fā)射過(guò)數(shù)據(jù)，那么這個(gè)時(shí)候?qū)?shù)組克隆管呵，然后調(diào)用 drain()方法梳毙。
void drain() {
...
v = ObjectHelper.requireNonNull(combiner.apply(s), "The combiner returned a null value");
...
a.onNext(v);
...
}
傳入的s就是上面數(shù)組保存的發(fā)射數(shù)據(jù)，a是觀察者捐下，即訂閱了combineLatest消息的實(shí)例顿天。
Join
任何時(shí)候，只要在另一個(gè)Observable發(fā)射的數(shù)據(jù)定義的時(shí)間窗口內(nèi)蔑担，這個(gè)Observable發(fā)射了一條數(shù)據(jù)，就結(jié)合兩個(gè)Observable發(fā)射的數(shù)據(jù)咽白。由
ObservableJoin
實(shí)現(xiàn)啤握。
- source
第一個(gè)被觀察者 - other（ObservableSource）
第二個(gè)參入合并的被觀察者。 - leftEnd（Function）
選擇當(dāng)前發(fā)射的每個(gè)項(xiàng)目的持續(xù)時(shí)間的函數(shù)晶框，用于判斷重疊的部分排抬。 - rightEnd（Function）
選擇當(dāng)前發(fā)射的每個(gè)項(xiàng)目的持續(xù)時(shí)間的函數(shù)，用于判斷重疊的部分授段。 - resultSelector（BiFunction）
一個(gè)函數(shù)蹲蒲，該函數(shù)計(jì)算任意兩個(gè)對(duì)象的結(jié)果。
subscribeActual方法
@Override
protected void subscribeActual(Observer<? super R> observer) {
JoinDisposable<TLeft, TRight, TLeftEnd, TRightEnd, R> parent =new JoinDisposable<>(observer, leftEnd, rightEnd, resultSelector);
// 通知觀察者侵贵，并傳入parent參數(shù)
observer.onSubscribe(parent);
LeftRightObserver left = new LeftRightObserver(parent, true);
// disposables到JoinDisposable集合中
parent.disposables.add(left);
LeftRightObserver right = new LeftRightObserver(parent, false);
// disposables到JoinDisposable集合中
parent.disposables.add(right);
// 調(diào)用第一個(gè)被觀察者的subscribe方法
source.subscribe(left);
// 第二個(gè)加入的被觀察者subscribe方法
other.subscribe(right);
}
調(diào)用第兩個(gè)（left和right）被觀察者的subscribe
source.subscribe(left);
會(huì)執(zhí)行LeftRightObserver中的onNext方法
public void onNext(Object t) {
parent.innerValue(isLeft, t);
}
插入數(shù)組隊(duì)列queue
queue.offer(isLeft ? LEFT_CLOSE : RIGHT_CLOSE, index);
-
執(zhí)行drain()方法
取出數(shù)據(jù)隊(duì)列第一個(gè)數(shù)據(jù)
Integer mode = (Integer)q.poll();如果讀取到數(shù)據(jù)是null届搁，者進(jìn)入下一次循環(huán)。
讀取第二個(gè)數(shù)據(jù)窍育，即我們發(fā)射的數(shù)據(jù)卡睦。
Object val = q.poll();-
mode == LEFT_VALUE
說(shuō)明是左邊的數(shù)據(jù)，即第一個(gè)被觀察者發(fā)射的數(shù)據(jù)漱抓。然后進(jìn)行類型轉(zhuǎn)換表锻。左邊的數(shù)據(jù)索引(leftIndex)自加，然后將該數(shù)據(jù)放進(jìn)該索引下的名為lefts的map集合中乞娄。
調(diào)用leftEnd的apply對(duì)數(shù)據(jù)進(jìn)行包裝瞬逊，返回ObservableSource實(shí)例。
創(chuàng)建LeftRightEndObserver實(shí)例仪或，將該實(shí)例加入JoinDisposable的disposables中确镊。
LeftRightEndObserver end = new LeftRightEndObserver(this, true, idx); disposables.add(end);調(diào)用數(shù)據(jù)包裝實(shí)例的subscribe并傳入JoinDisposable實(shí)例。
p.subscribe(end);接下來(lái)遍歷rights集合溶其，調(diào)用resultSelector并傳入兩個(gè)參與發(fā)射的數(shù)據(jù)骚腥。
R w = Objects.requireNonNull(resultSelector.apply(left, right)通知觀察者
a.onNext(w); mode == RIGHT_VALUE與
和mode == LEFT_VALUE順序相反。mode == LEFT_CLOSE
從lefts或者rights移除該索引下的數(shù)據(jù)瓶逃，disposables中移除該索引下的數(shù)據(jù)束铭，
-
LeftRightEndObserver
上面會(huì)調(diào)用數(shù)據(jù)包裝實(shí)例的subscribe方法廓块，那么在LeftRightEndObserver中做了什么操作呢？
public void onNext(Object t) {
if (DisposableHelper.dispose(this)) {
parent.innerClose(isLeft, this);
}
}
首先結(jié)束訂閱關(guān)系契沫，調(diào)用innerClose方法带猴。
public void innerClose(boolean isLeft, LeftRightEndObserver index) {synchronized (this) {
queue.offer(isLeft ? LEFT_CLOSE : RIGHT_CLOSE, index);}
drain();
}
那么又回到上面的邏輯了。
使用舉例
private void join(){Observable.<Integer>create(emitter -> emitter.onNext(1))
.join(observer -> observer.onNext(2), new Function<Integer, ObservableSource<Integer>>() {
@Override
public ObservableSource<Integer> apply(Integer integer) {
return observer -> {
// 立即調(diào)用懈万，發(fā)射的數(shù)據(jù)會(huì)被移除拴清，最后觀察者得不到任何數(shù)據(jù)
// observer.onComplete();
};
}
}, integer -> (ObservableSource<Integer>) observer -> {
// 立即調(diào)用，發(fā)射的數(shù)據(jù)會(huì)被移除会通，最后觀察者得不到任何數(shù)據(jù)
// observer.onComplete();
}, (BiFunction<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) { }
@Override
public void onNext(@NonNull Integer integer) {
// 輸出為：create operate--->join: 3
System.out.println("create operate--->join: " + integer);
}
@Override
public void onError(@NonNull Throwable e) { }
@Override
public void onComplete() { }
});
}
總結(jié)
LeftRightEndObserver作用就是將lefts口予，rights，disposables中對(duì)應(yīng)索引的數(shù)據(jù)移除涕侈，也就是數(shù)據(jù)有效的時(shí)間期限沪停，一旦被移除，就不能被通知給觀察者了裳涛。其實(shí)功能就是將兩個(gè)被觀察者（left和right）發(fā)射的數(shù)據(jù)由resultSelector（BiFunction）做變換后通知給觀察者木张。
比如你在p.subscribe(end)方法后，會(huì)執(zhí)行l(wèi)eftEnd或者rightEnd的subscribe方法端三，如果你這時(shí)立即調(diào)用LeftRightEndObserver的onNext或者onComplete方法舷礼，最終都是得不到resultSelector變換后的數(shù)據(jù)的。
所以為什么說(shuō)leftEnd和rightEnd是持續(xù)時(shí)間的函數(shù)郊闯。我們可以控制數(shù)據(jù)的有效時(shí)間妻献。
Merge
合并多個(gè)Observables的發(fā)射物。太多了团赁，自己查看源碼旋奢。如果想知道大概用法請(qǐng)參考：Merge
StartWith
在數(shù)據(jù)序列的開(kāi)頭插入一條指定的項(xiàng)。
Switch
將一個(gè)發(fā)射多個(gè)Observables的Observable轉(zhuǎn)換成另一個(gè)單獨(dú)的Observable然痊，后者發(fā)射那些Observables最近發(fā)射的數(shù)據(jù)項(xiàng)
Zip
通過(guò)一個(gè)函數(shù)將多個(gè)Observables的發(fā)射物結(jié)合到一起至朗，基于這個(gè)函數(shù)的結(jié)果為每個(gè)結(jié)合體發(fā)射單個(gè)數(shù)據(jù)項(xiàng)。下面講解一下ObservableZip實(shí)現(xiàn)剧浸，看一下里面的drain()方法實(shí)現(xiàn)
public void onNext(T t) {
queue.offer(t);
parent.drain();
}
public void drain() {
...
for (;;) {
for (;;) {
int i = 0;
int emptyCount = 0;
for (ZipObserver<T, R> z : zs) {
if (os[i] == null) {
boolean d = z.done;
T v = z.queue.poll();
boolean empty = v == null;
if (checkTerminated(d, empty, a, delayError, z)) {
return;
}
if (!empty) {
os[i] = v;
} else {
emptyCount++;
}
} else {
if (z.done && !delayError) {
Throwable ex = z.error;
if (ex != null) {
cancelled = true;
cancel();
a.onError(ex);
return;
}
}
}
i++;
}
if (emptyCount != 0) {
break;
}
R v;
try {
v = Objects.requireNonNull(zipper.apply(os.clone()), "The zipper returned a null value");
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
cancel();
a.onError(ex);
return;
}
a.onNext(v);
Arrays.fill(os, null);
}
missing = addAndGet(-missing);
if (missing == 0) {
return;
}
}
}
精華邏輯全在者幾層for循環(huán)里面锹引，由里向外，第一層for循環(huán)：遍歷observers唆香，而observers的長(zhǎng)度也是Observable參數(shù)個(gè)數(shù)嫌变，即有多少個(gè)發(fā)射的數(shù)據(jù)的Observable實(shí)例，而row保存發(fā)射的值的數(shù)組躬它。
如果Observable發(fā)射過(guò)數(shù)據(jù)腾啥，那么對(duì)應(yīng)取出來(lái)的值就不等于null，執(zhí)行else，如果沒(méi)有發(fā)生錯(cuò)誤或者終止倘待，那么繼續(xù)循環(huán)疮跑。
如果Observable沒(méi)有發(fā)射過(guò)數(shù)據(jù)，者將數(shù)據(jù)保存到對(duì)應(yīng)的數(shù)據(jù)索引下凸舵。
第二層循環(huán)祖娘，if (emptyCount != 0)為true，者跳出第二次循環(huán)啊奄，否則將row保存的數(shù)據(jù)應(yīng)用變換函數(shù)渐苏，然后把變換后的值發(fā)射給觀察者。
emptyCount是在row數(shù)組中還有Observable沒(méi)有發(fā)射過(guò)數(shù)據(jù)的時(shí)候菇夸，會(huì)加1琼富。
第三層循環(huán)，很簡(jiǎn)單庄新，自己查看公黑。
如果在地一層循環(huán)里面 ，f (os[i] == null)始終不成立摄咆，也就是說(shuō)所以O(shè)bservable都發(fā)射過(guò)數(shù)據(jù)了，那么調(diào)用變換函數(shù)人断，然后發(fā)射變換后的值吭从。
