angular2 學習筆記 ( rxjs 流 )

更新: 2019-11-24

startWith 和 pairwise 

s.pipe(
  startWith('c'),
  map(v => v + 1),
  tap(v => console.log(v)), // c1
  startWith(null),
  pairwise(),
  tap(v => console.log(v)), // [null, c1]
).subscribe(([before, after]) => {
  // console.log(before, after);
});

2 個點要留意

第一，pairwise 須要 2次值纔會觸發. 因此 startWith('c') 到 pairwise 就被吃掉了. subscribe 不會觸發

因此須要 startWith(null) 來餵它一次. 

第二, startWith 的次序. 最後一個 tap 的值是 [null, c1] 而不是 [c1, null] 

startWith('c1'),
startWith('c2'),
tap(v => console.log(v)), // [c2, c1]

要記住哦。

 

 

 

 

 

更新: 2019-07-18

unsubscribe vs complete 

在用 ng 的時候, 咱們會糾結何時要調用 unsubscribe, 由於聽說 angular 會幫咱們處理...

其實最好是每一次都調用 unsubscribe. 好比下面這個例子, setimeout 表明 component destroy 

當 destroy 時，即便我 subject.complete(), 我也沒法阻止 tap 的觸發. 因此仍是 unsubscribe 穩當一些.

按邏輯講, 當 component destroy 咱們是取消咱們對外部的監聽, 意思是咱們再也不處理了, 而 complete 則是它再也不發送了. 

它再也不發送, 和我再也不處理是 2 個概念. 它不發, 可是我手上的工做仍是得作完, 我不處理是我立馬停掉手上工做. 

const s = new Subject();
const o = s.asObservable();
const sub = o.pipe(delay(3000), tap(() => console.log('tap'))).subscribe(() => {
  console.log('done');
});
s.next('dada');
setTimeout(() => {
  s.complete();
  // sub.unsubscribe();
}, 1000);

 

 

更新 : 2019-06-2

defer 用於延後一個 function 執行. 當 subscribe 後才被執行, 一般用於作 promise retry 

好比我有一個 promise 方法

async function getValueAsync(ok: boolean): Promise<string> {
    return new Promise((resolve, reject) => {
        console.log('run');
        setTimeout(() => {
            if (ok) {
                resolve('dada');
            }
            else {
                reject('fail');
            }
        }, 3000);
    });
}

from(getValueAsync(false)).pipe(
    retry(1)
).subscribe({
    next: (value) => { console.log(value); },
    error: (err) => { console.log('Error: ' + err); },
    complete: () => { console.log('complete'); }
});

若是我直接這樣跑是不能 retry 的. refer : https://stackoverflow.com/questions/33072512/rx-frompromise-and-retry 這裏有解釋 

用 defer

defer(() =>  getValueAsync(false)).pipe(
    retry(2)
).subscribe({
    next: (value) => { console.log(value); },
    error: (err) => { console.log('Error: ' + err); },
    complete: () => { console.log('complete'); }
});

這樣就能夠了

 

 

 

更新 : 2018-03-12 

學 rxjs 最好的就是看官網的文檔，解釋的很清楚. 

http://cn.rx.js.org/manual/overview.html#h39

https://rxjs-cn.github.io/learn-rxjs-operators/

function subscribe(observer) {
  var intervalID = setInterval(() => {
    observer.next('hi');
  }, 1000);

  return function unsubscribe() {
    clearInterval(intervalID);
  };
}
var unsubscribe = subscribe({next: (x) => console.log(x)});
var unsubscribe = subscribe({next: (x) => console.log(x)});
// 稍後：
unsubscribe(); // 清理資源

上面這一段代碼讓我明白了幾個重點

1. 惰性 (當 observable 被建立時，沒有 subscribe 是不會開始運行的, 由於 observable 就像函數調用 )

2. 每個 subscribe 是獨立的 

3. observable 和經常使用的 addEventListener 不一樣，它不會吧全部的 observer 保存在列表裏頭. 

 

而 subject 則是道道地地的 addEventListener, 它會保存全部 observer 在列表裏. 

而把它們結合的方式就是 observable.subscribe(subject); 

由於 subject 就是一個 observer 擁有 next 方法. 

這就是 rxjs 其中 2 大核心, observable and subject 

至於其它的 subject 還有一堆的 operator, create observable 等，都是基於這 2 個核心的擴展而已. 

 

 

更新 : 2017-11-14 

最近重新看了 30 天 rxjs, 這裏補上一些筆記.

順便提一下, ng 5.x 開始 rxjs 的寫法換掉了 

參考 : https://github.com/ReactiveX/rxjs/blob/master/doc/lettable-operators.md

import { combineLatest } from 'rxjs/observable/combineLatest';
import { catchError, take } from 'rxjs/operators';

combineLatest(pendingEmitters).pipe(
    take(1),
    catchError(() => {
        reject();
        return '';
    })
).subscribe(() => {
    resolve();
});

不像 jquery 那樣串連了. 

 

1.concat 

https://ithelp.ithome.com.tw/articles/10187520

concat 是把多個 observeable 合併起來, 其特點是隻有前面一個 observeable complete 了後邊的 observeable 纔開始起做用.

concat(o1,o2,o3), o1 沒有 complete 的話, o2 怎麼叫都不會觸發.

另外一個說法就是 concat 先 subcribe o1, 而後等到 o1 completed 後再去 subscribe o2, 一直到完, 那它本身也就 completed 了.

concat 是能夠調用的 import { concat } from 'rxjs/observable/concat';

 

2. merge

https://ithelp.ithome.com.tw/articles/10187520

它和 concat 都是用來合併的. 區別是它不須要等 complete, 任何一個 observable 觸發都會有效果. 

換句話說就是 merge 會直接 subcribe 全部的 obserable 不像 concat 那樣會等 completed.

merge(o1,o2) o1 沒有 complete, o2 叫同樣有效.

 

3.concatAll

concatAll 屬於 operators 

o1.pipe(map(_ => o2),concatAll()) 

每一次 o1 叫，都會產生多一個 o2 

好比 o1 叫了 3 次 , 那麼就有 3 個 o2 

concatAll 就是把這 3 個組合起來( 每一次 o1 叫, 都會 push 新的 o2 去這個 array ) 而後 concat(o2,o2,o2)，後續的步驟就和 concat 一摸同樣了. 因此它起到了打平和 concat 的做用.

 

4. mergeAll

mergeAll 和 concat 是同一個原理只是, 最後不是用 concat 而是用 merge(o2,o2,o2);

另外, mergeAll(2) 能夠傳入一個變量去控制容許併發的數量 

好比你輸入 2,  那麼 a 叫了 3次, 你有 merge(o2,o2,o2), 第3個 o2 叫的時候原本是會有效果的，可是因爲限制了 2, 那麼只能等第一或第二個 o2 complete, 第 3 o2 叫纔有效果了。確保同一時期只有 2 個.

note mergeAll(1) === concatAll()

 

5.switchAll

沒有 switch 只有 switchAll 

它和 concatAll, mergeAll 的區別是, o1 每一次叫, 永遠只保留最新的 o2, 以前的 o2 通通丟掉. 

 

concatAll, switchAll, mergeAll 

https://ithelp.ithome.com.tw/articles/10188325

這 3 個都是用來打平 obs 

concatAll 上面講了重點是會等上一個 complete 纔去下一個 

switchAll 則是一旦有新的一個來，舊的就忽略掉. 

mergeAll 則是併發處理.

 

6 concatMap, switchMap, mergeMap 

https://ithelp.ithome.com.tw/articles/10188387

就是 map() + switchAll(), map() + concatAll(), map() + mergeAll() 的縮寫而已. 

還有個好處是它能夠傳入一個方法 (a,b,c,d) => next, 能夠獲取到 o1,o2 的值而後返回下一個值. 

 

6.5 exhaustMap 

這個和 concatMap 很像. 惟一的區別是, concat 等待第一個 complete 了之後會去 subscribe 下一個(第二個)

而 exhasust 呢, 它會去 subscribe 下下下下一個 (最後一個), 

 

7. combineLatest 

https://ithelp.ithome.com.tw/articles/10187638

它也是用來作合拼處理的,

它須要等 "每個" observable 至少有一個開始值以後纔開始工做. 這和 merge 不一樣, merge 不須要等

它每一次觸發均可以獲取到全部 observable 當前的值, 這和 merge 不一樣, merge 每一次觸發只能獲取一個 observable 的值.

 

8.withLatestFrom

https://ithelp.ithome.com.tw/articles/10187638

它和 combineLatest 同樣，惟一的區別是, 它只有 main observable next 值時纔會觸發 callback , 其它 observable next 只是記入值而已. 

 2019-06-14 補上一個例子, form value update 可是隻有 button click 的時候才 emit

this.formGroup = this.formBuilder.group({
  date: [null, Validators.required]
});

const f = this.formBuilder.group({
  name: [''],
  age: [11]
});
const button = new Subject();

button.pipe(withLatestFrom(f.valueChanges)).subscribe(v => {
  console.log(v);
});


f.setValue({ name: 'dada', age: 15 });
f.setValue({ name: 'dada', age: 18 });
button.next('dada');
button.next('dada2');

 

 

9. zip 

https://ithelp.ithome.com.tw/articles/10187638

它也是合併. 

它的關鍵是順位, 好比 2 個 observables, 2 個都 next 1次 的時候就會 callback 並獲得 2 個的第一個值, 若是 2 個不平均, 好比一個 next 了 10 次, 另外一個 next 2 次, 那麼 callback 就只有 2 次. 

等第 2 個 next 第 3 次時, callback 就會獲得 第1和第2個的第3次 next 的值. 

 

10. scan 

https://ithelp.ithome.com.tw/articles/10187882

就是 js 的 reduce, 區別在於它老是返回 observable. 每一次 next 觸發, callback 均可以獲取上一次的值 + 如今的值作一些處理, 返回下一個值. 

 

11. buffer 

https://ithelp.ithome.com.tw/articles/10187882

buffer, bufferTime, bufferCount 

它用於累積 next 等待另外一個 obserable next 的時候才觸發 callback 

o1.pipe(buffer(o2))...   o2 next 的時候 o1 的 callback 才觸發, 而且返回期間全部的 o1 next 值. 

 

12. delay & delayWhen

https://ithelp.ithome.com.tw/articles/10187999

delay(1000) 就是延遲 1 秒咯, 若是咱們要每一次都不用的話. 

就用 delayWhen(v => empty().delay(v + 1000)); 必須返回一個 observable

 

13.debounce & debounceTime 

https://ithelp.ithome.com.tw/articles/10188121

它和 buffer 有點像, 都是會累積值, 可是區別在於, 當一個新值被 next 進來, 它會把以前的值釋放掉, 而且時間重新開始算. 

 

14. throttle & throttleTime

https://ithelp.ithome.com.tw/articles/10188121

它用於限制一個時間內, 最高觸發的頻率. 好比 throttleTime(1000) 就限制了一秒內無論 next 幾回, 只有第一次會 callback 日後的都不會, 直到下一秒開始. 

 

14.1 auditTime

它和 debounceTime 很像，只是 debounceTime 會 clear 掉上一次，而這個不會

3個分別的使用 : 

用戶 keydown 時我想監聽 

1. debounceTime 1000， 用戶連續 keydown 我等 1 秒, 一秒中內用戶又 keydown 了，我從新記時，再等 1 秒... 一直到用戶 1 秒內再也沒有 keydown 我才觸發

2. auditTime 1000 用戶連續 keydown, 我等 1 秒，一秒中內用戶又 keydown 了, 但我 "不" 從新記時了, 1 秒後我就觸發, 也就是說, 用戶 1 秒內按多少次，我都當成 1 次 而且在 1 秒後才觸發. 

3. debounceTime 1000, 用戶連續 keydown, 我直接觸發, 一秒中內用戶又 keydown 了, 我不理，直到 1 秒後 

特點

debounceTime 從新幾時, 一直不觸發

auditTime 等..觸發

debounceTime 觸發...等

 

 

15.distinct, distinctUntilChanged

https://ithelp.ithome.com.tw/articles/10188194

它就是咱們熟悉的 distinct, 若是值相同就忽略 callback 

distinct((x) => { return x.value }); 能夠提供一個取值的方法，對比估計是用 === 

distinct 會把全部的值都存起來作對比, distinctUntilChanged 的區別是它只會存最後一次的值作對比.

 

16.catchError, retry, retryWhen, repeat

https://ithelp.ithome.com.tw/articles/10188263

catchError 是捕獲錯誤, 

catchError((error, obs) => obs); 返回第 2 個參數 obs 能夠實現從跑. 

retry 就是作了上述的事情, 而 retry 多了一個能夠指定次數, retry(3) 從試 3 次 

retryWhen(errorObs => errorObs.delay(1000)) retryWhen 可讓咱們操做更多，好比間隔多久才 retry 下一次等等. 

repeat 和 retry 是同樣的，區別在於 retry 必須在 error 時纔會有, 而 repeat 則是無論有沒有 error 都會執行. 

 

 

17.  ReplaySubject, BehaviorSubject

behavior 表明一個保持值的 subject, 一旦訂閱立刻會觸發 callback 並獲取到最新的值. 即使不訂閱也能夠調用 .value 來獲取當前值.

replay 會緩存以前的 next 一旦新的訂閱加入，就會 playback 以前全部的 next 值. 

 

18. race 

race(s1, s2, s3) 的意思是, 哪個先觸發，那麼以後我就 watch 這一個罷了，另外 2 個 subject 就不理會了。

 

19. mapTo

對比 map, mapTo 的參數是一個值，而不是獲取值的方法, 因此值就只有一個，要留意哦。

 

 

  

rxjs 在處理 dom 事件時是很是好用的. 

步驟通常上是 獲取全部 element, 創建全部的 event 

而後就是各作 rxjs operator 對 event 和 element 的處理. 

參考 : https://ithelp.ithome.com.tw/articles/10187756

 

 

 

 

  

更新 : 2017-10-14 

import 'rxjs/add/observable/combineLatest'; //每個至少一次後纔開始觸發(無需 completed), 一個一次能夠獲取你們最新的 value

import 'rxjs/add/observable/forkJoin'; // 相似 promise.all 全部 observable 須要 最少next 一次 and completed 纔會觸發,因此只觸發一次獲取全部 value

import 'rxjs/add/observable/merge'; // 用來監聽多個 click event 一個一次, merge(a,b) 若是 a,b 都有初始值, 那麼會立刻觸發 2 次.

import 'rxjs/add/observable/concat'; // 和 merge 同樣, 惟一不一樣的是它須要第一個 completed 纔會觸發第 2 個, 第一還沒 completed 第 2 next 都不會叫哦

 

更新 2017-05-17 

今天被 toPromise 給騙了. 

我一直覺得, 全部的 "流" 均可以輕鬆的轉成 await stream.toPromise();

後來我發現有個流一直沒有反應 

let subject = new BehaviorSubject('a');
let o = subject.asObservable();
o.toPromise().then(() => console.log('pro')); //不會跑
setTimeout(()=> {
  subject.next('haha');
  //subject.complete();
}, 1000);

上網找了一下才發現，原來 toPromise().then 必須是 completed 纔會跑

因此上面的 subject.complete() 必需要打開才行. 這也意味着 toPromise 只能用在一次的 async 中, 若是是要持續 subscribe 的狀況下請使用 .subscribe() 

 

 

 

更新 2017-04-02 

Subject 的主要功能就是觀察者模式. 

咱們能夠隨時寫入值，徹底本身操控. 

可是有時候咱們但願它依賴於其它 stream 那麼咱們使用 connect 

let s1 = new Subject();
let o1 = s1.asObservable(); //咱們想以來的 stream 

let s2 = new Subject();
s2.subscribe(v => console.log(v));

o1.subscribe(v => s2.next(v),v => s2.error(v)); //第1種寫法,超麻煩
o1.subscribe(s2); //第2種,但是我沒有要立刻 subscribe 的話呢 ?  
let connector = o1.multicast(s2); //第3種
connector.connect();

s1.next("value");

 

 

更新 2017-03-31

好文,力推 : http://ithelp.ithome.com.tw/articles/10189028?sc=iThomeR
再談談 cold & hot 

observeable 是 default cold 的. 

code 的意思是說, 當有多個 subscribe 時，每個都是一條獨立的鏈.

好比 http 多個 subscribe 的話，你會發現你的 request 會發了好幾個.

hot 的意思則是每一個 subscribe 共享一個鏈, 無論你什麼以後插入subscribe 你都不會重新開始. 

把一個 cold 變成 hot 的方法是使用 Subject 充當中間人. 

具體看這 3 篇就明白了 

http://ithelp.ithome.com.tw/articles/10188633

http://ithelp.ithome.com.tw/articles/10188677

http://ithelp.ithome.com.tw/articles/10188750

這裏介紹一下 ReplaySubject

Subject.subscirbe() ， 不會立刻執行, 由於要等待下一個 Subject.next

BehaviorSubject.subscribe() ， 立刻執行, 由於裏面必定會有值. 

ReplaySubject.subscribe(), 不必定立刻執行，若是曾經 .next 過纔會執行 

multicast, refCount,publish,share 的目的就是把 cold 轉換成 hot .

其原理就是使用了 Subject 系列. 

multicast 後來被 publish 取代了. publish 對應 subject 因此有 publishBehavior, pulishReplay 

refCount 是 connect 的意思，就是把 observer 連接上 subject 的動做. 

因爲 publish().refCount() 太常常用到，因此發明了 share 寫的更快了嘻嘻。

在使用 ng 的 http 時要注意. observable 是 cold 的。但不少狀況下咱們更但願它是 hot 的. 

每一次的 subscribe 應該只返回同一個結果, 而這個 http 只發一次請求. 

這時咱們須要這樣寫 : http.get().publishReplay(1).refCount() 

publishReplay(1) 以後的每個 subscribe 都會獲得同一個資料了.

 

 

更新 : 2017-03-27 

何時須要 unsubscribe ? 

http 不須要, router param 也不須要. 

下面說說幾個狀況 

1. Subject.complete 以後, 全部的 subscribe 都不會再觸發, 新的 subscribe 也加不進來了. 

因此若是咱們知道訂閱的 Subject 以後會被 complete 那麼咱們能夠無需擔憂 unsubscribe 的問題 

2. 使用 async/await toPromise 能夠避開 unsubscribe 的問題. 

3. 可使用 .first(判斷) 表示何時開始拿而後停 (好比一個值須要等待 ajax)

4. 使用 takeWhile(判斷) 來決定何時取消訂閱. 

5. 使用 OnDestroy 

 

 

更新 : 2017-03-18

Observable, 

Subject,

BehaviorSubject

的區別和用法 

 

當咱們手中有一個 Observable, 咱們能夠去監聽它，那麼日後的事情發生咱們都會知道。

可是, 以前發生的事情咱們都沒辦法知道. 咱們也沒辦法用它廣播一個新的事件。

Observable 能作的事情, Subject 均可以作到. 而 Subject 多了一個能力，就是能夠用它廣播一個新事件出去. 

因此若是你站在一個監聽者的角度, Subject 和 Observable 沒啥區別. 你依然沒辦法知道過往發生的事情. 也只有在下一次事件廣播時纔會被通知. 

Subject 能作的事情 BehaviorSubject 均可以作，能監聽也能廣播. 最大的特色在於它能知道過往的事情。

當你手中有一個 BehaviorSubject 你能夠立刻調用 .value 獲取當前的值, 你訂閱它的話，你也會立馬收到一個事件，而不像 Subject 或 Observable 一直傻傻等下一次廣播才能獲得值. 

 

本身選擇用吧.

 

ng 的 http.get 返回的是 Observable, 你一監聽它你會立刻得到響應, 從這裏咱們就能夠推斷出沿着這條鏈往上走最終就是一個 BehaviorSubject. 由於只有 BehaviorSubject 被監聽的時候纔會立刻獲得響應。

 

2016-09-23

RxJS 博大精深，看了好幾篇文章都沒有明白. 

範圍牽扯到了函數響應式開發去了... 我對函數式只知其一;不知其二, 響應式更是第一次聽到... 

唉...不過日子仍是得過...混着過先唄

我目前所理解的很淺, 大體上是這樣的概念.

1.某些場景下它比 promise 好用, 它善於過濾掉不關心的東西. 

2.它是觀察者模式 + 迭代器模式組成的 

3.跟時間，事件, 變量有密切關係

4.世界上有一種東西叫 "流" stream, 一個流能表示了一段時間裏，同樣東西發生的變化. 

  好比有一個值, 它在某段時間裏從 "我" 變成 "你" 再變成 "他". 

  而咱們能夠對這個流進行觀察，因此只要它發生變化，咱們就會發現而後作任何事情。

5.站在遊覽器的角度, 服務器推送數據過來, 用戶操做界面, timer 都是咱們關心的流.

好，來看例子. 

咱們經過 new Subject 來建立流. 也可使用 new EventEmitter 或者 BehaviorSubject. 這些都繼承了 Subject

EventEmitter 是 ng2 提供的

BehaviorSubject 能夠填入初始值

import { Subject } from "rxjs/Subject";
private textEmitter: Subject<string> = new Subject(); 

要改變流中的值，咱們使用 .next(value), 這個是迭代器的概念咯

keyup(value : string)
{
    this.textEmitter.next(value);
}

那麼訂閱是這樣的 

ngOnInit() {
    this.text$ = this.textEmitter
        .debounceTime(500)
        .distinctUntilChanged()
        .switchMap(v => this.getDataAsync(v));

    this.text$.subscribe((value) => {
        console.log(value);
    });            
}

input keyup 性能優化, 咱們一般會寫一個 timeout + cleartimeout 的方式, 這個 debounceTime 就是幹這個的 

流更新結束後 500ms 纔會通知觀察者 

distinctUntilChanged 是說只有當值和上一次通知時的值不同的時候才通知觀察者 

.map 和 .switchMap 都是用來對值進行處理的, 這個和 array.map 概念是同樣的

而 .map 和 .switchMap 的區別是 .swichMap 處理那些返回 Observeable 的值 

getDataAsync(value : string): Observable<string>
{        
    let subject = new Subject();
    setTimeout(() => {
        console.log("after 2second");
        subject.next(value + "final");
    }, 2000);
    return subject;
}

若是咱們使用 map 的話，它會直接返回 "subject" 這個對象, 而若是用 switchMap 它會返回這個 subject 對象的響應值.

<input type="text" #input (keyup)="keyup(input.value)" />
<p>{{ text$ | async }}</p>

ng2 提供了一個 async Pipe, 它會監聽左邊這個 text$ stream. 後面加一個 $ 符號一般用來代表這是一個 stream.

還有一個經常使用的功能是 combineLatest

就是能夠同時監聽多個流，只要其中一個有變更，那麼全部的最新值都會發布出去, 能夠用來實現依賴屬性.

這裏須要注意一點 combineLatest 的全部流都必須有值, 不能夠是一個歷來都沒有 next 過的 Observable 否則它就不會運行了.

最簡單的方法是使用 observable.startWith(null) 讓它有一個值. 

@Component({
    selector: "compute-property",
    template: ` 
        <input type="text" #input1 (keyup)="text1.next(input1.value)" />
        <input type="text" #input2 (keyup)="text2.next(input2.value)" />  
        {{ result$ | async }}                 
    `
})
export class ComputePropertyComponent implements OnInit {

    text1: BehaviorSubject<string> = new BehaviorSubject<string>("a");
    text2: BehaviorSubject<string> = new BehaviorSubject<string>("b");
    result$: Observable<string>;
    constructor() {}
    
    ngOnInit() {
        this.result$ = Observable.combineLatest(this.text1, this.text2).map(values => {          
            return values[0] + " " + values[1];
        }); 
    }     
}

還有 bufferCount, bufferTime 也是經常使用到

text: Subject<number> = new Subject<number>();
    
ngOnInit() {
    this.text.bufferCount(2)
        .subscribe(v => console.log(v)); //[v1,v2] 存夠 count 了就發佈

    this.text.bufferTime(2000)
        .subscribe(v => console.log(v)); //[v1,v2,...]把 2 秒內的全部 next value 放進來
}

Observable.of 能夠簡單的返回一個默認值 

Observable.of<string>("").subscribe(v => console.log(v));

rxjs 整個文檔很是大，要按需加載.

一般作法是爲項目開一個 rxjs-operators.ts 

import 'rxjs/add/observable/throw'; 
import 'rxjs/add/observable/combineLatest'; 
import 'rxjs/add/observable/from'; 
import 'rxjs/add/observable/of'; 
import 'rxjs/add/operator/catch';
import 'rxjs/add/operator/debounceTime';
import 'rxjs/add/operator/distinctUntilChanged';
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/switchMap';
import 'rxjs/add/operator/toPromise';
import 'rxjs/add/operator/startWith'; 
import 'rxjs/add/operator/bufferCount';
import 'rxjs/add/operator/bufferTime';

放入經常使用的方法 

而後再 app.module.ts 裏面導入它 

import './rxjs-operators'; 

 

Hot or cold , share or not

refer : 

http://blog.thoughtram.io/angular/2016/06/16/cold-vs-hot-observables.html

http://blog.csdn.net/tianjun2012/article/details/51351823

1. by default, observable is not share.

let sub = new Subject();
let obs = sub.map(v => {
    console.log("ajax call"); 
});
obs.subscribe(v => console.log("subscribe 1"));
obs.subscribe(v => console.log("subscribe 2"));         
sub.next("value");

ajax 發了 2 次. angular2 的 Http 也是 not share 哦. 

因此當咱們有多個 subscribe 的時候要想想是否咱們須要 share 

let obs = sub.map(v => {
    console.log("ajax call"); 
}).share();

調用一個 share 方法就能夠了，或者是 

let obs = sub.map(v => {
    console.log("ajax call"); 
}).publish().refCount();

效果是同樣的. 

 

by default, observable is cold.

意思是說只有在 subscribe 出現了之後纔會啓動. ( 當第一個 subscribe 出現時, observable 就會馬上啓動了哦 ) 

let sub = new Subject();
let obs = sub.map(v => {
    console.log("ajax call");
});
sub.next("aaa");
//obs.subscribe(v => console.log("subscribe 1"));
//obs.subscribe(v => console.log("subscribe 2")); 

ajax 不會觸發. 

若是咱們但願它在沒有 subscribe 的狀況下觸發的話, 能夠這樣寫. 

let sub = new Subject();
let obs = sub.map(v => {
    console.log("ajax call");
}).publish();
obs.connect();
sub.next("aaa");

至於什麼狀況下使用哪種，我尚未實戰，之後再說.

多一個例子解釋: 

let obs = Observable.create(observer => {
    console.log("observer run");
    observer.next(Date.now());
});
obs.subscribe(v => console.log("1st subscriber: " + v));
obs.subscribe(v => console.log("2nd subscriber: " + v));     
//observer run
//1st subscriber: 1474649902498
//observer run
//2nd subscriber: 1474649902501 

no share. 因此 observer run 了 2 次. 

let obs = Observable.create(observer => {
    console.log("observer run");
    observer.next(Date.now());
}).share();
obs.subscribe(v => console.log("1st subscriber: " + v));
obs.subscribe(v => console.log("2nd subscriber: " + v));     
//observer run
//1st subscriber: 1474650049833

share 了, 因此 observer only run 1 次.

cold, 因此當第一個 subcribe 出現後 observer 馬上運行 -> .next 更新了 value -> 第一個 subcribe callback 被調用 -> 整個過程結束 -> 而後第2個 subcribe 註冊 .. 因爲是 share 因此 observer 沒有載被觸發. 第2個 subscribe callback 沒有被調用. 

延後觸發的作法 : 

let obs = Observable.create(observer => {
    console.log("observer run");
    observer.next(Date.now());
}).publish();
obs.subscribe(v => console.log("1st subscriber: " + v));
obs.subscribe(v => console.log("2nd subscriber: " + v));
obs.connect();
//observer run 
//1st subscriber: 1474650370505 
//2nd subscriber: 1474650370505

能夠看到 .publish() 以後, subscribe 再也不能激活 observer 了，而必須手動調用 .connect() 才能激活 observer. 

這幾個例子只是爲了讓你瞭解它們的玩法.

小結:

observer default is cold and not share.

cold 表示只有 subscribe 出現 observer 纔會被激活.

not share 表示每個 subscribe 都會激活 observer 鏈. 

 

經常使用 : 

1. finally 的使用

import 'rxjs/add/operator/finally';

this.http.get(
    "http://localhost:58186/api/products",
    { headers: new Headers({ "Accept": "application/json" })}
).finally(() => {
    console.log("finally"); //無論 success or error 最後都會跑這個
}).subscribe(response => {        
    console.log("success");
}, response => {
    console.log("fail");
}, () => {
    console.log("success final");
});
//result : 
//success -> success final -> finally
//fail -> finally

 

2. 錯誤處理 throw catch  

import 'rxjs/add/operator/catch';
import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/throw';

this.http.get(
    "http://localhost:58186/api/products",
    {
        headers: new Headers({ "Accept": "application/json" })
    })
    .map(r => r.json())
    .catch((r) => {
        if ("1" == "1") {
            //do something ...
            return null; //catch 了在返回真確
        }
        else {
            return Observable.throw("error"); //catch 了繼續返回錯誤
        }
    })
    .subscribe(
        r => console.log(r),
        r => { console.log("fail") }
    );

 

3. previous / current value 

用 .pairwise()

let userSubject = new BehaviorSubject<string>("default value");
let user$ = userSubject.asObservable().pairwise();
user$.subscribe(([before, after]) => { console.log(before), console.log(after); });  
userSubject.next("super");
userSubject.next("ttc");
//result : 
//["default value","super"]
//["super","ttc"]