Outline
[TOC]
前言
大家喜聞樂見的Backpressure來啦.
這一節(jié)中我們將來學(xué)習(xí)Backpressure. 我看好多吃瓜群眾早已坐不住了, 別急, 我們先來回顧一下上一節(jié)講的Zip.
正題
上一節(jié)中我們說到Zip可以將多個(gè)上游發(fā)送的事件組合起來發(fā)送給下游, 那大家有沒有想過一個(gè)問題, 如果其中一個(gè)水管A發(fā)送事件特別快, 而另一個(gè)水管B 發(fā)送事件特別慢, 那就可能出現(xiàn)這種情況, 發(fā)得快的水管A 已經(jīng)發(fā)送了1000個(gè)事件了, 而發(fā)的慢的水管B 才發(fā)一個(gè)出來, 組合了一個(gè)之后水管A 還剩999個(gè)事件, 這些事件需要繼續(xù)等待水管B 發(fā)送事件出來組合, 那么這么多的事件是放在哪里的呢? 總有一個(gè)地方保存吧? 沒錯(cuò), Zip給我們的每一根水管都弄了一個(gè)水缸 , 用來保存這些事件, 用通俗易懂的圖片來表示就是:
如圖中所示, 其中藍(lán)色的框框就是zip給我們的水缸! 它將每根水管發(fā)出的事件保存起來, 等兩個(gè)水缸都有事件了之后就分別從水缸中取出一個(gè)事件來組合, 當(dāng)其中一個(gè)水缸是空的時(shí)候就處于等待的狀態(tài).
題外話: 大家來分析一下這個(gè)水缸有什么特點(diǎn)呢? 它是按順序保存的, 先進(jìn)來的事件先取出來, 這個(gè)特點(diǎn)是不是很熟悉呀? 沒錯(cuò), 這就是我們熟知的隊(duì)列, 這個(gè)水缸在Zip內(nèi)部的實(shí)現(xiàn)就是用的隊(duì)列, 感興趣的可以翻看源碼查看.
好了回到正題上來, 這個(gè)水缸有大小限制嗎? 要是一直往里存會怎樣? 我們來看個(gè)例子:
Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
for (int i = 0; ; i++) { //無限循環(huán)發(fā)事件
emitter.onNext(i);
}
}
}).subscribeOn(Schedulers.io());
Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("A");
}
}).subscribeOn(Schedulers.io());
Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
@Override
public String apply(Integer integer, String s) throws Exception {
return integer + s;
}
}).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.d(TAG, s);
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Log.w(TAG, throwable);
}
});
在這個(gè)例子中, 我們分別創(chuàng)建了兩根水管, 第一根水管用機(jī)器指令的執(zhí)行速度來無限循環(huán)發(fā)送事件, 第二根水管隨便發(fā)送點(diǎn)什么, 由于我們沒有發(fā)送Complete事件, 因此第一根水管會一直發(fā)事件到它對應(yīng)的水缸里去, 我們來看看運(yùn)行結(jié)果是什么樣.
運(yùn)行結(jié)果GIF圖:
我勒個(gè)草, 內(nèi)存占用以斜率為1的直線迅速上漲, 幾秒鐘就300多M , 最終報(bào)出了OOM:
zlc.season.rxjava2demo W/art: Throwing OutOfMemoryError "Failed to allocate a 28 byte allocation with
4194304 free bytes and 8MB until OOM;
zlc.season.rxjava2demo W/art: "main" prio=5 tid=1 Runnable
zlc.season.rxjava2demo W/art: | group="main" sCount=0 dsCount=0 obj=0x75188710 self=0x7fc0efe7ba00
zlc.season.rxjava2demo W/art: | sysTid=32686 nice=0 cgrp=default sched=0/0 handle=0x7fc0f37dc200
zlc.season.rxjava2demo W/art: | state=R schedstat=( 0 0 0 ) utm=948 stm=120 core=1 HZ=100
zlc.season.rxjava2demo W/art: | stack=0x7fff971e8000-0x7fff971ea000 stackSize=8MB
zlc.season.rxjava2demo W/art: | held mutexes= "mutator lock"(shared held)
zlc.season.rxjava2demo W/art: at java.lang.Integer.valueOf(Integer.java:742)
出現(xiàn)這種情況肯定是我們不想看見的, 這里就可以引出我們的Backpressure了, 所謂的Backpressure其實(shí)就是為了控制流量, 水缸存儲的能力畢竟有限, 因此我們還得從源頭去解決問題, 既然你發(fā)那么快, 數(shù)據(jù)量那么大, 那我就想辦法不讓你發(fā)那么快唄.
那么這個(gè)源頭到底在哪里, 究竟什么時(shí)候會出現(xiàn)這種情況, 這里只是說的Zip這一個(gè)例子, 其他的地方會出現(xiàn)嗎? 帶著這個(gè)問題我們來探究一下.
我們讓事情變得簡單一點(diǎn), 從一個(gè)單一的Observable說起.
來看段代碼:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
for (int i = 0; ; i++) { //無限循環(huán)發(fā)事件
emitter.onNext(i);
}
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Thread.sleep(2000);
Log.d(TAG, "" + integer);
}
});
這段代碼很簡單, 上游同樣無限循環(huán)的發(fā)送事件, 在下游每次接收事件前延時(shí)2秒. 上下游工作在同一個(gè)線程里, 來看下運(yùn)行結(jié)果:
哎臥槽, 怎么如此平靜, 感覺像是走錯(cuò)了片場.
為什么呢, 因?yàn)樯舷掠喂ぷ髟?code>同一個(gè)線程呀騷年們! 這個(gè)時(shí)候上游每次調(diào)用emitter.onNext(i)其實(shí)就相當(dāng)于直接調(diào)用了Consumer中的:
public void accept(Integer integer) throws Exception {
Thread.sleep(2000);
Log.d(TAG, "" + integer);
}
所以這個(gè)時(shí)候其實(shí)就是上游每延時(shí)2秒發(fā)送一次. 最終的結(jié)果也說明了這一切.
那我們加個(gè)線程呢, 改成這樣:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
for (int i = 0; ; i++) { //無限循環(huán)發(fā)事件
emitter.onNext(i);
}
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Thread.sleep(2000);
Log.d(TAG, "" + integer);
}
});
這個(gè)時(shí)候把上游切換到了IO線程中去, 下游到主線程去接收, 來看看運(yùn)行結(jié)果呢:
可以看到, 給上游加了個(gè)線程之后, 它就像脫韁的野馬一樣, 內(nèi)存又爆掉了.
為什么不加線程和加上線程區(qū)別這么大呢, 這就涉及了同步和異步的知識了.
當(dāng)上下游工作在同一個(gè)線程中時(shí), 這時(shí)候是一個(gè)同步的訂閱關(guān)系, 也就是說上游每發(fā)送一個(gè)事件必須等到下游接收處理完了以后才能接著發(fā)送下一個(gè)事件.
當(dāng)上下游工作在不同的線程中時(shí), 這時(shí)候是一個(gè)異步的訂閱關(guān)系, 這個(gè)時(shí)候上游發(fā)送數(shù)據(jù)不需要等待下游接收, 為什么呢, 因?yàn)閮蓚€(gè)線程并不能直接進(jìn)行通信, 因此上游發(fā)送的事件并不能直接到下游里去, 這個(gè)時(shí)候就需要一個(gè)田螺姑娘來幫助它們倆, 這個(gè)田螺姑娘就是我們剛才說的水缸 ! 上游把事件發(fā)送到水缸里去, 下游從水缸里取出事件來處理, 因此, 當(dāng)上游發(fā)事件的速度太快, 下游取事件的速度太慢, 水缸就會迅速裝滿, 然后溢出來, 最后就OOM了.
這兩種情況用圖片來表示如下:
同步:
異步:
從圖中我們可以看出, 同步和異步的區(qū)別僅僅在于是否有水缸.
相信通過這個(gè)例子大家對線程之間的通信也有了比較清楚的認(rèn)知和理解.
源頭找到了, 只要有水缸, 就會出現(xiàn)上下游發(fā)送事件速度不平衡的情況, 因此當(dāng)我們以后遇到這種情況時(shí), 仔細(xì)思考一下水缸在哪里, 找到水缸, 你就找到了解決問題的辦法.
既然源頭找到了, 那么下一節(jié)我們就要來學(xué)習(xí)如何去解決了. 下節(jié)見.
