摘要： # rxjs簡單入門 > rxjs全名Reactive Extensions for JavaScript，Javascript的響應式擴展, 響應式的思路是把隨時間不斷變化的數據、狀態、事件等等轉成可被觀察的序列(Observable Sequence)，然后訂閱序列中那些Observable對象的變化，一旦變化，就會執行事先安排好的各種轉換和操作 **rxjs適用於異步場景，即前端

rxjs簡單入門

rxjs全名Reactive Extensions for JavaScript，Javascript的響應式擴展, 響應式的思路是把隨時間不斷變化的數據、狀態、事件等等轉成可被觀察的序列(Observable Sequence)，然后訂閱序列中那些Observable對象的變化，一旦變化，就會執行事先安排好的各種轉換和操作

rxjs適用於異步場景，即前端交互中接口請求、瀏覽器事件以及自定義事件。通過使用rxjs帶給我們前所未有的開發體驗。

1. 統一異步編程的規范，不管是Promise、ajax還是事件，通通封裝成序列(Observable Sequence)，一旦有異步環節發生變更，觀察序列即可截獲發生變更的信息。
2. 前端業務層和展現層解耦，比如展現層不需要關系指定事件觸發時和DOM無關的處理邏輯。同時業務層也能組裝異步操作中多個異步邏輯之間的關系，無需暴露給展現層。展現層關心的是：異步操作其中環節的數據變化。

3. rxjs開發業務層具有高彈性，高穩定性，高實時性等特點。

   廢話不多說，此篇文檔結合模擬場景的例子，通過傻瓜式的描述來說明rxjs常用的方法以及組合關系。

1. Let's Go

rxjs應用觀察者模式，其中包含2個重要的實例：Observer觀察者和Subject被觀察對象，多個Observer注冊到Subject中，在Subject功能觸發時，會通知注冊好的Observab列表，逐一通知其響應觀察變更信息。

1.1 quick start

1. 先從官網搬來rxjs的幾個實例概念
   • Observable: 可觀察的數據序列.
   • Observer: 觀察者實例，用來決定何時觀察指定數據.
   • Subscription: 觀察數據序列返回訂閱實例.
   • Operators: Observable的操作方法，包括轉換數據序列，過濾等，所有的Operators方法接受的參數是上一次發送的數據變更的值，而方法返回值我們稱之為發射新數據變更.
   • Subject: 被觀察對象.
   • Schedulers: 控制調度並發，即當Observable接受Subject的變更響應時，可以通過scheduler設置響應方式，目前內置的響應可以調用Object.keys(Rx.Subject)查看。
2. 我們最常用也最關心的Observable，四個生命周期：創建 、訂閱 、 執行 、銷毀。
   • 創建Obervable，返回被觀察的序列源實例，該實例不具備發送數據的能力，相比之下通過new Rx.Subject創建的觀察對象實例具備發送數據源的能力。
   • 通過序列源實例可以訂閱序列發射新數據變更時的響應方法（回調方法）
   • 響應的動作實際上就是Observable的執行
   • 通過序列源實例可以銷毀，而當訂閱方法發生錯誤時也會自動銷毀。
   • 序列源實例的catch方法可以捕獲訂閱方法發生的錯誤，同時序列源實例可以接受從catch方法返回值，作為新的序列源實例

3. 掌握最簡單的例子

4. // 5.0.0-rc.1
   import Rx from 'rxjs';
   //emit 1 from promise
   const source = Rx.Observable.fromPromise(new Promise(resolve => resolve(1)));
   //add 10 to the value
   const example = source.map(val => val + 10);
   //output: 11
   const subscribe = example.subscribe(val => console.log(val));
   

   　　通過代碼掌握Observable, Observer, Subscription, Operators, Subject和Schedulers之間的關系

   import Rx from 'rxjs';
   /**
     Rx.Observable是Observable
     Rx.Observable.create創建序列源source，創建source的方法有多個，比如of, from, fromPromise等
     observer是Observer觀察者，只有在Rx.Observable.create創建方法可以獲取，其他創建方法內置了observer且不可訪問
     observer.next發射數據更新
     source.map其中map就是Operators的其中一個方法，方法調用返回新的source1
     source1.subscribe是訂閱，即數據更新時的響應方法。同時返回訂閱實例Subscription
     subscription.next立即響應（不同於發射）靜態數據，此時不會經過`Operators`處理
     ! Rx.Observable.create或者Rx.Subject.create創建的source不會自動關閉，其他方式則當檢測到沒有序列發生變更會自動銷毀source.
   */
   const source = Rx.Observable.create(observer => {
     observer.next('foo');
     setTimeout(() => observer.next('bar'), 1000);
   });
   const source1 = source.map(val => `hello ${val}`);
   const subscription = source1.subscribe(value => console.log(value));
   subscription.next('foo1');
   
   // forEach和subscribe相似，同是實現訂閱效果，等到promise可以監控subscription完成和失敗的異常。
   // 日志打印並沒有comlete, 因為source並沒有完成關閉，觸發調用observer.complete()
   const promise = source1.forEach(value => console.log(value))
   promise.then(() => console.log('complete'), (err) => console.log(err));
   /**
     output: 
     hello foo
     foo1
     hello foo
     hello bar
     hello bar
   */
   
   /**
     new Subject創建被觀察者實例，同source一樣都具備subscribe方法，表示的含義和作用也一樣，即發射數據變更時響應方法。
     subject.next立即發射數據變更，作用同observer.next
     注意foo1是最后輸出的，是因為在創建source時指定了Rx.Scheduler.async，是異步的調度器，表示在響應數據處理時是異步執行的。
   */
   Rx.Observable.of('foo1', Rx.Scheduler.async).subscribe(value => console.log(value));
   
   const subject = new Subject();
   const source2 = subject.map(val => `hello ${val}`);
   const subscription = source1.subscribe(value => console.log(value));
   subject.next('foo');
   subscription.next('bar');
   /**
     output: 
     hello foo
     bar
     foo1
   */
   

   　　

   1. 1.2 學會看rxjs交互圖

      交互圖中每條連表示一個數據序列，每個球表示每次發射的變更，最后一條線表示最終產出的數據序列。

   下圖以combineLastest來舉例：

   • 方法之上的每條線都是一個source(數據序列實例)
   • 方法之下方法調用后返回的新source
   • combineLastest表示被組合的每個source，一旦發射數據變更，必須拿到其余的source的最新值（當異步時則等待，直到都拿到最新值），組合為新的數據，作為新source發射的數據變更。source1: ————————①——————————②——————————③————————————④—————————⑤——————————|——> source2: ———————————ⓐ————————ⓑ————————————ⓒ—————————————————————ⓓ—————————|——> combineLastest(source1, source2, (x, y) => x + y) source: ———————(①ⓐ)—(②ⓐ)—(②ⓑ)—————(③ⓑ)—(③ⓒ)———(④ⓒ)————(⑤ⓒ)—(⑤ⓓ)——|——>

   2. 實例方法Operators

   前面講過Operators方法調用時，接收的參數是source，返回新的source, 以下是個人學習使用過程中，簡單總結的rxjs各方法用法。

   2.1 創建

   • 發射完數據更新自動關閉：from, fromPromise, of, from, range
   • 不發射直接關閉：empty
   • 拋出異常后關閉：throw
   • 不發射數據也不關閉：never
   • 保持發射數據且不自動關閉：timer, interval, fromEvent
   • 需要手動發射數據且不自動關閉：create, (還有Rx.Subject.create)

   2.2 轉換

   1. 1:1效果：map, mapTo, flatMap, scan, expand, pluck
      • map，source = source1.map(func)表示source1每次發射數據時經過func函數處理，返回新的值作為source發射的數據
      • mapTo，不同於map，func改為靜態值
      • flatMap，當發射的數據是一個source時，在訂閱的響應方法中接收到的也是一個source（這是合理的，發射什么數據就響應什么數據嘛，但是如果我們想在響應方法收到的是source的發射數據），flatMap就是可以允許發射數據是一個source，同時在響應的時候接收的是source的發送數據，后面我們稱之為**source打平**
      • scan，source = source1.scan(func, initialValue), source每次發射的數據是source前次發射數據和source1當前發射的數據 的組合結果（取決於func，一般是相加), initialValue第一次發射，source前次沒發射過，采用initialValue作為前次發射的數據
      • expand，和scan不同的是當func返回值是一個source時，在func接收到的數據是source打平后的發射數據。**特別適用於polling長輪詢**
      • pluck，每次發射數據時，獲取數據中的指定屬性的值作為source的發射數據
   2. 1:N效果：concat, concatAll, concatMap, concatMapTo, merge, mergeAll, mergeMap, mergeMapTo, switchMap,switchMapTo
      • concat, concatAll和merge, mergeAll屬於組合類型，放在這講更好體現其效果。
      • concat，source = source1.concat(source2)表示source發射數組的順序是，當source1或source2發射數據，source就發射。但是只有當source1發射完且關閉(source1不在發送數據)后，才觸發source2發射數據。
      • concatAll，不同於concat，會把所有的發射的數據打平（如果數據為source時），然后在決定下次發射哪個數據。
      • concatMap，source = source1.concatMap(source2)表示source1每次發射數據時，獲取source2的所有發射數據，map返回多個待發射數據，按順序發射第一個數據變更。
      • concatMapTo, 不同於concatMap, map處理以source2的數據為返回結果
      • switchMap, 和concatMap不同的是在map之后的待發射數據排序上，concatMap中source1每次發射時source2的所有發射數據都接收，作為source1下一次發射前，之間的所有發射數據。switchMap則會判斷source2的所有發射數據是否有數據的發射時間比source1下一次發射的時間晚，找出來去除掉。
      • switchMapTo對switchMap就好比concatMap對concatMapTo, mergeMap對比mergeMapTo的關系也是如此。
      • mergeMap相比於switchMap，找出的數據會打平到source中，不丟棄。
   3. N:1效果：buffer, bufferCount, bufferTime, bufferWhen
      • buffer，source = source1.buffer(source2)表示source1以source2為參考，在source2的2次發射數據之間為時間段，source才發射一次數據，數據為該時間段內source1本該發射的數據的組合。
      • 比如source1原先每隔1秒發射一次數據，source2是每個2秒發射數據，source = source1.buffer(source2), 那么source會每隔2秒發射數據（source1的2秒內發射的2個數值組成的數組）
      • bufferCount，source = source1.bufferCount(count, start), count表示source1毎3次發射數據作為source的一次發射數據，發射完后，以source1當前組合的發射數據的第start個開始算下次發射數據需要組合的起始數據。
      • bufferTime，一段時間內的source1發射數據作為source的一次發射數據
      • bufferWhen, 以默認結果為准分成2段，分別作為source的每次發射數據
   4. 1:source效果：groupBy, window, windowCount, windowTime, windowWhen
      • groupBy, source = source1.groupBy(func), 表示source1的所有發射數據，按func分成多段，每段作為source的每次發送的數據（這里數據只是新的source，你可以理解為inner Observable實例)
      • window和buffer不同的時，source每次發送的是innerObservable
      • window vs windowCount vs windowTime vs windowWhen 同 buffer相似
   5. 1:sources效果：partition
      • partition，sources = source1.partition(func), 根據func吧所有的source1發射數據分段，每段組成一個source，最終得到sources數組

   2.3 過濾

   source的過濾不會對發射數據做任何改變，只是減少source的發射次數，所以理解起來會簡單很多，這里只做個簡單分類

   • 防抖動（一段時間內只取最新數據作為一次發射數據，其他數據取消發射）：debounce, debounceTime,throttle(和debounce唯一區別是debounce取一段時間內最新的，而throttle忽略這段時間后，發現新值才發送）,throttleTime
   • 去重（重疊的發射數據只去第一數據作為發射數據，其他相同數據取消發射）：distinct, distinctUntilChanged
   • 定位（根據條件值去一個或部分幾個數據作為對應發射數據，其他取消發射）：elementAt, first, last, filter,take, takeLatst, takeUntil, takeWhile,
   • 跳過（根據條件去除符合條件的，取剩下的值作為每次發射數據）：skip, skipUntil, skipWhile,ignoreElements(忽略所有的，等同於empty)
   • 樣本：sample, source=source1.sample(source2), 以source2發射數據時來發現最新一次source1發射的數據，作為source的發射數據，個人覺得應該屬於**轉換**分類，官網放到了**過濾**

   2.4 組合

   做個source組合成新的souce

   • concat, concatAll和merge, mergeAll，在**轉換**分類講過了
   • combineLastest，source = source1.combineLastest(source2, func)，source1和source2一旦發射數據，func會觸發，拿到source1和source2最新的發射數據，返回新的數據，作為source的發射數據。
   • combineAll，同combineLastest，，source = sources.combineAll()
   • forkJoin，source = Rx.Observable.forkJoin(sources), 所有的sources都關閉后，獲取各自最新的發射數組組合為數組，作為source的發射數據
   • zip和forkJoin的區別是，zip是sources都有發送數據時，組合為一個數組作為source的發送數據，而sources任一source關閉了，則取source最后發射的數值。
   • zipAll，同concat對concatAll
   • startWith，source = source1.startWith(value), 表示在source1的最前面注入第一次發射數據
   • withLastestFrom, soruce = source1.withLastestFrom(source2, func), 表示source1每次發射數據時，獲取source2最新發射的數據，如果存在則func處理得到新的數組作為source的發射數據

   2.5 判斷

   • find和findIndex分別是指定發射數據和發射數據的下標（第幾次發送的），應該放到**過濾**分類才合理
   • isEmpty, every, include等，判斷是否為真，判斷的結果當做是source的發射數據

   2.6 錯誤處理

   • catch，source在Operators調用過程中出現的異常，都可以在catch捕獲到，同時可以返回新的source，因為出現異常的當前source會自動銷毀掉。
   • retry，source = source.retry(times), source的所有發射，重復來幾遍。
   • retryWhen，根據條件來決定來幾遍，只有當條件為false時才跳出循環。

   2.7 工具

   • do，在每次響應訂閱前，可以通過source.do(func)，做一些提前處理等任何動作，比如打印一下發射的數據等。
   • delay, delayWhen，每次發送數據時，都延遲一定時間間隔后再發送。
   • observeOn, 設置scheduler,即發射數據的響應方式，Schedulers詳細查看地址, 這里不講解了，項目中應用得不多。
   • subcribeOn, timeInterval設置sheduler
   • toPromise, source轉成promise，可以通過promise.then達到source.subscribe的效果
   • toArray，把source所有發射的數據，組成數組輸出。

   2.8 計算

   把source的所有發射數據進行指定計算后，得出的數據作為新source的發射數據，計算方法分別有：max, min, count,reduce, average等

   2.9 其他

   • cache, source = source1.cache(1);共享source1的訂閱結果，即不管source訂閱幾回，響應方法接收到的發射數據都是同一份。
   • 共享source訂閱結果很重要，因為**組合**等方法組合多個source時，其中包含sourceA，同時sourceA還需要單獨訂閱其結果，在不用cache情況下，sourceA會產生2個subscription，即2個訂閱實例，但是我們更希望是能達到sourceA發生變化時，都能通知到所有的組合sourceA的source。
   • publish，publishSource = source.publish(),讓source的訂閱的工作延后，即source不會發射數據，而是等到publishSource.connect()調用后才開發發射數據。效果和delay很相似，不同的是可以控制合適發射。
   • share，當source訂閱多次，那么每次響應時do都會調用多次，通過share合並響應，則source發射一次數據更新，多次響應當當一次響應處理，do也調用一次。

   參考資料

   1. rxjs官網 - http://reactivex.io/rxjs/
   2. rxjs代碼 - https://github.com/ReactiveX/rxjs
   3. 常用rxjs方法的交互圖 - http://rxmarbles.com/
   4. rxhjs教程 - http://xgrommx.github.io/rx-book/content/observable/observable_instance_methods/toarray.html
   5. Scheduler - https://mcxiaoke.gitbooks.io/rxdocs/content/Scheduler.html
   6. 原文地址：https://yq.aliyun.com/articles/65027