rxjs簡單入門

時間 2019-11-12

標籤 rxjs 簡單入門简体版

原文原文鏈接

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

rxjs簡單入門

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

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

統一異步編程的規範，不論是Promise、ajax仍是事件，統統封裝成序列(Observable Sequence)，一旦有異步環節發生變動，觀察序列便可截獲發生變動的信息。
前端業務層和展示層解耦，好比展示層不須要關係指定事件觸發時和DOM無關的處理邏輯。同時業務層也能組裝異步操做中多個異步邏輯之間的關係，無需暴露給展示層。展示層關心的是：異步操做其中環節的數據變化。
rxjs開發業務層具備高彈性，高穩定性，高實時性等特色。java

廢話很少說，此篇文檔結合模擬場景的例子，經過傻瓜式的描述來講明rxjs經常使用的方法以及組合關係。react

1. Let's Go

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

1.1 quick start

先從官網搬來rxjs的幾個實例概念
- Observable: 可觀察的數據序列.
- Observer: 觀察者實例，用來決定什麼時候觀察指定數據.
- Subscription: 觀察數據序列返回訂閱實例.
- Operators: Observable的操做方法，包括轉換數據序列，過濾等，全部的Operators方法接受的參數是上一次發送的數據變動的值，而方法返回值咱們稱之爲發射新數據變動.
- Subject: 被觀察對象.
- Schedulers: 控制調度併發，即當Observable接受Subject的變動響應時，能夠經過scheduler設置響應方式，目前內置的響應能夠調用Object.keys(Rx.Subject)查看。
咱們最經常使用也最關心的Observable，四個生命週期：建立、訂閱、執行、銷燬。
- 建立Obervable，返回被觀察的序列源實例，該實例不具有發送數據的能力，相比之下經過new Rx.Subject建立的觀察對象實例具有發送數據源的能力。
- 經過序列源實例能夠訂閱序列發射新數據變動時的響應方法（回調方法）
- 響應的動做實際上就是Observable的執行
- 經過序列源實例能夠銷燬，而當訂閱方法發生錯誤時也會自動銷燬。
- 序列源實例的catch方法能夠捕獲訂閱方法發生的錯誤，同時序列源實例能夠接受從catch方法返回值，做爲新的序列源實例
掌握最簡單的例子github
```
// 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之間的關係ajax
```
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