一、觀察者模式
觀察者模式（Observer）完美的將觀察者和被觀察的對(duì)象分離開崔拥。舉個(gè)例子，用戶界面可以作為一個(gè)觀察者拆魏，業(yè)務(wù)數(shù)據(jù)是被觀察者，用戶界面觀察業(yè)務(wù)數(shù)據(jù)的變化渤刃，發(fā)現(xiàn)數(shù)據(jù)變化后贴膘，就顯示在界面上卖子。面向?qū)ο笤O(shè)計(jì)的一個(gè)原則是：系統(tǒng)中的每個(gè)類將重點(diǎn)放在某一個(gè)功能上，而不是其他方面刑峡。一個(gè)對(duì)象只做一件事情洋闽，并且將他做好。觀察者模式在模塊之間劃定了清晰的界限突梦，提高了應(yīng)用程序的可維護(hù)性和重用性诫舅。
兩種觀察者模式：
- Observable ( 被觀察者 ) / Observer ( 觀察者 )
-
Flowable （被觀察者）/ Subscriber （觀察者）
觀察者模式
背壓(backpressure):
對(duì)于可丟棄的事件,上游生產(chǎn)過快導(dǎo)致事件堆積，當(dāng)堆積到超出buffer緩沖區(qū)上限,就叫做Backpressure出現(xiàn)
在rxjava中會(huì)經(jīng)常遇到一種情況就是被觀察者發(fā)送消息太快以至于它的操作符或者訂閱者不能及時(shí)處理相關(guān)的消息宫患。那么隨之而來的就是如何處理這些未處理的消息骚勘。
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()) //上游在io線程
.observeOn(AndroidSchedulers.mainThread()) //下游在主線程
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Thread.sleep(2000);
Log.d(TAG, "" + integer);
}
});
有兩種避免這種情況的方法：
- 1:從數(shù)量上進(jìn)行治理, 減少發(fā)送的事件
比如按一定條件過濾出需要的事件
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
for (int i = 0; ; i++) {
emitter.onNext(i);
}
}
}).subscribeOn(Schedulers.io())
.filter(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return integer % 10 == 0;//只取能整除的部分
}
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "" + integer);
}
});
//或者暴力一點(diǎn)，用onBackpressureDrop撮奏，如果消費(fèi)者無法處理數(shù)據(jù)，則它就把該數(shù)據(jù)丟棄了当宴。
Observable.interval(1, TimeUnit.MILLISECONDS)
.onBackpressureDrop()
.observeOn(Schedulers.newThread())
.subscribe(
i -> {
System.out.println(i);
try {
Thread.sleep(100);
} catch (Exception e) { }
},
System.out::println);
- 2:從速度上進(jìn)行治理, 減緩事件發(fā)送的速度
比如：操作符中 sample() 畜吊、 throttleLast(?)、 throttleFirst(?)户矢、throttleWithTimeout(?)玲献、debounce(?) 可以通過調(diào)節(jié)速率來改變Observable發(fā)射消息的速度。
建立“響應(yīng)式拉動(dòng)（reactive pull）”backpressure
也就是觀察者調(diào)用request(n)來向被觀察者告知自己的事件處理能力梯浪，從而控制被觀察者發(fā)送事件的數(shù)量
二捌年、基本使用
①.被觀察者（Observable）的創(chuàng)建
//被觀察者，假如這是數(shù)不到3的G胖
Observable halflife = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("半條命1出啦");
emitter.onNext("半條命2出啦");
emitter.onNext("沒啦");
emitter.onComplete();
emitter.onNext("半條命3出啦挂洛！");//嗯哼
}
});
//也可以是
Observable halflife = Observable.just("半條命1出啦","半條命2出啦","沒啦");
//或者
String[] words = {"半條命1出啦","半條命2出啦","沒啦"};
Observable halflife = Observable.from(words); //from還可以從callable礼预，publisher，future中獲取
//兩種都會(huì)依次調(diào)用
//emitter.onNext("半條命1出啦");
//emitter.onNext("半條命2出啦");
//emitter.onNext("沒啦");
//emitter.onComplete();
通過create()方法生成對(duì)象托酸，ObservableOnSubscribe<T>可以當(dāng)作計(jì)劃表，泛型T是要操作的對(duì)象励堡。重寫subscribe方法应结，里面寫具體計(jì)劃鹅龄。ObservableEmitter<String>對(duì)象中的Emitter是發(fā)射器的意思，有3種方法：void onNext(T value)漩绵、void onError(Throwable error)止吐、void onComplete()碍扔。onNext方法可以無限調(diào)用秕重，Observer（觀察者）所有的都能接收到溶耘，onError和onComplete是互斥的，Observer（觀察者）只能接收到一個(gè)百新，OnComplete可以重復(fù)調(diào)用饭望，但是Observer（觀察者）只會(huì)接收一次铅辞，而onError不可以重復(fù)調(diào)用萨醒，第二次調(diào)用就會(huì)報(bào)異常富纸。
注意：如果from()里面執(zhí)行了耗時(shí)操作，就算使用了subscribeOn(Schedulers.io())高职，也會(huì)是在主線程執(zhí)行怔锌，所以耗時(shí)操作最好還是用Observable.create(…);
? 除了create()，just(),from()方法創(chuàng)建以外還有
-
interval()—>定時(shí)發(fā)送 -
range(final int start, final int count)—>定時(shí)發(fā)送從start開始count次 -
repeat(long times)--- >指定要發(fā)多少次涝涤，要是times是空的話會(huì)無限次發(fā)送阔拳，相當(dāng)于repeat(Long.MAX_VALUE) -
defer--> 直到有觀察者訂閱才創(chuàng)建ovservable并且是給每個(gè)觀察者創(chuàng)建新的ovservable -
empty & never & error---> empty不發(fā)出數(shù)據(jù)但是正常終止糊肠，never不發(fā)數(shù)據(jù)也不終止(那有啥用啊货裹。精偿。笔咽。)，error的話不發(fā)數(shù)據(jù)但是會(huì)以錯(cuò)誤終止(onerror)
變換操作
Rxjava里面斯撮，將發(fā)送的事件或事件序列，加工后轉(zhuǎn)換成不同的事件或事件序列帕膜，也就是變換操作很常見垮刹。
map:對(duì)原始Observable發(fā)射的每一項(xiàng)數(shù)據(jù)應(yīng)用一個(gè)你選擇的函數(shù)，然后返回一個(gè)發(fā)射這些結(jié)果的Observable酪劫。默認(rèn)不在任何特定的調(diào)度器上執(zhí)行。如果指定observeOn則在observeon的那個(gè)線程上進(jìn)行刻剥，沒有指定observeon但是指定了scheduleon的話就在scheduleon的線程上造虏。都不指定就在observable的創(chuàng)建線程上執(zhí)行
- map 轉(zhuǎn)換是一對(duì)一的漓藕，原來發(fā)射了幾個(gè)數(shù)據(jù)享钞，轉(zhuǎn)換之后還是幾個(gè)
-
map 轉(zhuǎn)換可以改變發(fā)射的數(shù)據(jù)類型
Observable.create(
(ObservableOnSubscribe<Integer>) e -> {
e.onNext(1);
e.onNext(2);
e.onComplete();
}
).map(new Function<Integer, String>(){
@Override
public String apply(Integer integer) throws Exception {
String day;
switch (integer){
case 1: day = "半條命1";
break;
case 2: day = "半條命2";
break;
default: day = "G胖呢栗竖？";
break;
}
return day;
}
}).subscribe(new Observer<String>() {
Disposable mdisposable;
@Override
public void onSubscribe(Disposable d) {
mdisposable=d;
}
@Override
public void onNext(String s) {
Log.e(TAG,s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
Log.e(TAG,"Complete");
mdisposable.dispose();
}
});
flatMap:
flatMap()操作符的作用是將被觀察者發(fā)送的事件序列進(jìn)行拆分 & 單獨(dú)轉(zhuǎn)換 在合并成為一個(gè)新的事件序列最后在進(jìn)行發(fā)送
原理：將被觀察者發(fā)送的事件序列進(jìn)行拆分成一個(gè)個(gè)事件 在將每個(gè)事件都生成創(chuàng)建一個(gè)新的Observable對(duì)象
每個(gè)原始事件都會(huì)生成一個(gè)新的Observable對(duì)象
每個(gè)拆分的新的事件生成的新的Observable對(duì)象最終會(huì)匯總到一個(gè)新建總的Observable對(duì)象中
新建的總的Observable對(duì)象最終將新合并的事件序列發(fā)送給觀察者Observer
? 應(yīng)用場(chǎng)景:無序的將整個(gè)被觀察者發(fā)送的事件序列進(jìn)行變換(可能是我寫太少結(jié)果是有序的划滋。处坪。架专。)
咱去翻源碼看看吧部脚。。丧没。
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper,
boolean delayErrors, int maxConcurrency, int bufferSize) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
if (this instanceof ScalarCallable) {
@SuppressWarnings("unchecked")
T v = ((ScalarCallable<T>)this).call();
if (v == null) {
return empty();
}
return ObservableScalarXMap.scalarXMap(v, mapper);
}
return RxJavaPlugins.onAssembly(new ObservableFlatMap<T, R>(this, mapper, delayErrors, maxConcurrency, bufferSize));//可以看到這里生成了一個(gè)ObservableFlatMap
}
順著ObservableFlatMap走呕童，看它subscribeActual的實(shí)現(xiàn)
@Override
public void subscribeActual(Observer<? super U> t) {
if (ObservableScalarXMap.tryScalarXMapSubscribe(source, t, mapper)) {
return;
}
source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
}
也就是外部裝飾成一個(gè)MergeObserver夺饲，再看它的onNext實(shí)現(xiàn)
@Override
public void onNext(T t) {
// safeguard against misbehaving sources
if (done) {
return;
}
ObservableSource<? extends U> p;
try {
p = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null ObservableSource");
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
s.dispose();
onError(e);
return;
}
if (maxConcurrency != Integer.MAX_VALUE) {
synchronized (this) {
if (wip == maxConcurrency) {
sources.offer(p);
return;
}
wip++;
}
}
subscribeInner(p);
}
默認(rèn)調(diào)用的flatmap 的 maxConcurrency 大小是 Integer.MAX_VALUE擂找，所以最終會(huì)調(diào)用subscribeInner(p)贯涎，其中這個(gè)p是我們傳入的Function生成的Observable撼嗓。繼續(xù)往下走：
void subscribeInner(ObservableSource<? extends U> p) {
for (;;) {
if (p instanceof Callable) {
tryEmitScalar(((Callable<? extends U>)p));
if (maxConcurrency != Integer.MAX_VALUE) {
synchronized (this) {
p = sources.poll();
if (p == null) {
wip--;
break;
}
}
} else {
break;
}
} else {
InnerObserver<T, U> inner = new InnerObserver<T, U>(this, uniqueId++);
if (addInner(inner)) {
p.subscribe(inner);
}
break;
}
}
}
嗯且警，發(fā)現(xiàn)了傳給p的是InnerObserver斑芜，接著看它的onNext
@Override
public void onNext(U t) {
if (fusionMode == QueueDisposable.NONE) {
parent.tryEmit(t, this);
} else {
parent.drain();
}
}
默認(rèn)mode應(yīng)該是NONE，所以會(huì)執(zhí)行MergeObserable 的 tryEmit方法盈包，該方法中調(diào)用了drainLoop()
在這里面循環(huán)獲取數(shù)據(jù)并發(fā)射呢燥。
過濾操作:
可以用filter叛氨，elementAt & firstElement & lastElement寞埠，distinct等各種方法來對(duì)發(fā)送的事件進(jìn)行過濾
Observable.range(1,10).filter(i -> i > 5).subscribe(System.out::println);
Observable.range(1, 10).elementAt(0).subscribe(System.out::println);
②.觀察者的創(chuàng)建
Observer user = new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
mDisposable =d;
Log.e(TAG,"onSubscribe");
}
@Override
public void onNext(String s) {
Log.e(TAG,"onNext:"+s);
}
@Override
public void onError(Throwable e) {
Log.e(TAG,"onError"+e.getMessage());
mDisposable.dispose();
}
@Override
public void onComplete() {
Log.e(TAG,"complete");
mDisposable.dispose();
}
};
onNext()仁连、onError()阱穗、onComplete()都是跟被觀察者發(fā)射的方法一一對(duì)應(yīng)的揪阶，這里就相當(dāng)于接收了。Disposable英文意思是可隨意使用的，這里就相當(dāng)于讀者和連載小說的訂閱關(guān)系蕴茴，如果讀者不想再訂閱該小說了，可以調(diào)用 mDisposable.dispose()取消訂閱蒋畜。（如果在onNext里面dispose姻成，那么onError愿棋，onComplete都會(huì)收不到）
? 除了 Observer 接口之外糠雨，RxJava 還內(nèi)置了一個(gè)實(shí)現(xiàn)了 Observer 的抽象類：Subscriber。 Subscriber 對(duì) Observer 接口進(jìn)行了一些擴(kuò)展琅攘，但他們的基本使用方式完全一樣坞琴。(不過Subscriber相比較observer來說主要是多了onStart()和unsubscribe()兩個(gè)方法剧辐，用后者)
③.建立訂閱關(guān)系
halflife.subscribe(user);//一行代碼搞定(上面也有各種版本的)
輸出結(jié)果：
這樣就是rxjava最簡單的用法浙于。
RxJava的異步和鏈?zhǔn)骄幊?/h2>
RxJava是支持異步的挟纱，但是RxJava是如何做到的呢？這里就需要Scheduler（調(diào)度器）檀轨。它是RxJava用來控制線程参萄。當(dāng)我們沒有設(shè)置的時(shí)候煎饼，RxJava遵循哪個(gè)線程產(chǎn)生就在哪個(gè)線程消費(fèi)的原則，也就是說線程不會(huì)產(chǎn)生變化马篮，始終在同一個(gè)浑测。一般使用RxJava都是后臺(tái)執(zhí)行歪玲，前臺(tái)調(diào)用，本著這個(gè)原則岖圈，我們需要調(diào)用observeOn(AndroidSchedulers.mainThread())幅狮，observeOn是事件回調(diào)的線程株灸，AndroidSchedulers.mainThread()一看就知道是主線程慌烧，subscribeOn(Schedulers.io())，subscribeOn是事件執(zhí)行的線程厕氨，Schedulers.io()是子線程命斧，這里也可以用Schedulers.newThread()嘱兼，只不過io線程可以重用空閑的線程芹壕，因此多數(shù)情況下 io() 比 newThread() 更有效率。
//改一下上面的代碼
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("半條命1出啦");
emitter.onNext("半條命2出啦");
emitter.onNext("沒啦");
emitter.onComplete();
}
}).observeOn(AndroidSchedulers.mainThread())//回調(diào)在主線程
.subscribeOn(Schedulers.io())//執(zhí)行的子線程
.subscribe(new Observer<String>() {@Override
public void onSubscribe(Disposable d) {
mDisposable =d;
Log.e(TAG,"onSubscribe");
}
@Override
public void onNext(String s) {
Log.e(TAG,"onNext:"+s);
}
@Override
public void onError(Throwable e) {
Log.e(TAG,"onError"+e.getMessage());
mDisposable.dispose();
}
@Override
public void onComplete() {
Log.e(TAG,"complete");
mDisposable.dispose();
}
});
如果只用關(guān)心onNext方法里的內(nèi)容，可以直接重載subscribe(Consumer<? spuer T> onNext)這個(gè)方法睁壁，會(huì)減少代碼互捌，當(dāng)然一開始還是建議創(chuàng)建Observer對(duì)象疫剃。
線程調(diào)度—— Scheduler 與 Worker
Scheduler 與 Worker 在 RxJava2 中是一個(gè)非常重要的概念，他們是 RxJava 線程調(diào)度的核心與基石牲阁。
Rxjava2 Schedulers中默認(rèn)有多種shceduler的實(shí)現(xiàn)：
| 方法 | 說明 |
|---|---|
| single() | 每個(gè)worker都在同一個(gè)線程上執(zhí)行 |
| computation() | 適用于計(jì)算密集型任務(wù) |
| io() | 適用于 IO 密集型任務(wù) |
| trampoline() | 在調(diào)用schedule的線程執(zhí)行 |
| newThread() | 和single相反备燃，它每個(gè)worker都對(duì)應(yīng)一個(gè)新的線程 |
一個(gè) Scheduler 可以創(chuàng)建多個(gè) Worker并齐，這兩者
是一對(duì)多的關(guān)系客税，而 Worker 與 Task 也是一對(duì)多
的關(guān)系。
對(duì)于worker：同一個(gè)worker創(chuàng)建的task都是串行的测垛，執(zhí)行的任務(wù)符合隊(duì)列(先進(jìn)先出)秧均，worker會(huì)與調(diào)用它的方法的runnable綁定，如果worker取消了锯七，它的task也會(huì)都取消
因此當(dāng)有操作符需要使用 Scheduler 時(shí)眉尸，可以通過 Worker 來將一系列的 Runnable 統(tǒng)一的調(diào)度和取消巫延，最典型的例子就是observeOn，下面會(huì)嘗試分析炉峰。
/**
* Retrieves or creates a new {@link Scheduler.Worker} that represents serial execution of actions.
* <p>
* When work is completed it should be unsubscribed using {@link Scheduler.Worker#dispose()}.
* <p>
* Work on a {@link Scheduler.Worker} is guaranteed to be sequential.
*
* @return a Worker representing a serial queue of actions to be executed
*/
@NonNull
public abstract Worker createWorker();
@NonNull
public Disposable scheduleDirect(@NonNull Runnable run) {
return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
}
public Disposable schedulePeriodicallyDirect(@NonNull Runnable run, long initialDelay, long period, @NonNull TimeUnit unit)
Rxjava2中有scheduleDirect / schedulePeriodicallyDirect這兩個(gè)方法簡化對(duì)worker的調(diào)用疼阔，相當(dāng)于創(chuàng)建了一個(gè)只調(diào)度一次任務(wù)的worker并立即調(diào)度任務(wù)。
這里只挑io()說一下吧迅细。大致原理圖是這樣的
IoScheduler:
ioscheduler的線程數(shù)是沒有上限的茵典，因?yàn)?IO 設(shè)備的速度是遠(yuǎn)遠(yuǎn)低于 CPU 速度。等待io操作時(shí)cpu一般時(shí)閑置的彩倚。所以可以創(chuàng)建更多的線程來盡量充分利用cpu
CachedWorkerPool:
CachedWorkerPool是一個(gè)變長并定期清理的ThreadWorker的緩存池帆离，內(nèi)部通過一個(gè)ConcurrentLinkedQueue維護(hù)结澄。和PoolWorker類似，ThreadWorker也是繼承自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;
}
}
expirationTime表示超時(shí)時(shí)間们妥，在CacheWorkerPool初始化的時(shí)候會(huì)傳入worker超時(shí)時(shí)間所袁，這個(gè)時(shí)間表示ThreadWorker閑置后的最長存在時(shí)間(當(dāng)然餐曼，實(shí)際中也不一定有那么長)
EventLoopWorker
在ComputationScheduler中有EventLoopWorker這個(gè)類，它的作用主要是
- 管理自身調(diào)度過的任務(wù)
- ThreadWorker的管理集惋，回收和再次使用
Worker的管理
- 創(chuàng)建: 會(huì)現(xiàn)在隊(duì)列里找閑置的
ThreadWorker刮刑，有的話就取出养渴，沒有就會(huì)新建一個(gè)，在外面包裝一層EventLoopWorke并返回
+回收：dispose后翘紊，會(huì)把threadworker放進(jìn)閑置隊(duì)列并更新超時(shí)時(shí)間
+清除：CacheWorkerPool的定時(shí)任務(wù)帆疟，每隔一個(gè)超時(shí)時(shí)間(60s)清除隊(duì)列里的超時(shí)的threadworker
使用場(chǎng)景舉例
與retrofit聯(lián)用
先是retrofit的這幾個(gè):
private var okhttp = OkHttpClient.Builder()
.addInterceptor(
HttpLoggingInterceptor()
.setLevel(HttpLoggingInterceptor.Level.BODY))
.connectTimeout(8, TimeUnit.SECONDS)
.build()
private val client = Retrofit.Builder()
.baseUrl("https://news-at.zhihu.com/")
.client(okhttp)
.addConverterFactory(GsonConverterFactory.create())
.addCallAdapterFactory(RxJavaCallAdapterFactory.create())
.build()
private val api:Service = client.create(Service::class.java)
<uses-permission android:name="android.permission.INTERNET" />//別忘了這個(gè)(哭)
RetrofitUtil retro = new RetrofitUtil();
retro.getNews()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<Data>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Data data) {
mdata = data;
}
});
偷了點(diǎn)懶，不過簡單實(shí)現(xiàn)基本上是這樣自赔。
如有哪些地方說得不對(duì)還望指正
參考文章:
RxJava2 系列 -1：一篇的比較全面的 RxJava2 方法總結(jié)(WngShhng):http://www.reibang.com/p/823252f110b0
RxJava 2.0中backpressure(背壓)概念的理解(一葉飄舟):
https://blog.csdn.net/jdsjlzx/article/details/52717636
RxJava2源碼解讀之 Map绍妨、FlatMap(三好碼農(nóng))
http://www.reibang.com/p/122abe149ac4
深入理解 RxJava2：Scheduler（2）(蝶翼的罪)
http://www.reibang.com/p/b742526c7dec
使用Retrofit+RxJava實(shí)現(xiàn)網(wǎng)絡(luò)請(qǐng)求(JYcoder)
