1. 程式人生 > >構建流式應用—RxJS詳解

構建流式應用—RxJS詳解

       最近在 Alloyteam Conf 2016 分享了《使用RxJS構建流式前端應用》,會後在線上線下跟大家交流時發現對於 RxJS 的態度呈現出兩大類:有用過的都表達了 RxJS 帶來的優雅編碼體驗,未用過的則反饋太難入門。所以,這裡將結合自己對 RxJS 理解,通過 RxJS 的實現原理、基礎實現及例項來一步步分析,提供 RxJS 較為全面的指引,感受下使用 RxJS 編碼是怎樣的體驗。

目錄

  • 常規方式實現搜尋功能
  • RxJS · 流 Stream
  • RxJS 實現原理簡析
    • 觀察者模式
    • 迭代器模式
    • RxJS 的觀察者 + 迭代器模式
  • RxJS 基礎實現
    • Observable
    • Observer
  • RxJS · Operators
    • Operators ·入門
    • 一系列的 Operators 操作
  • 使用 RxJS 一步步實現搜尋功能
  • 總結

常規方式實現搜尋

做一個搜尋功能在前端開發中其實並不陌生,一般的實現方式是:監聽文字框的輸入事件,將輸入內容傳送到後臺,最終將後臺返回的資料進行處理並展示成搜尋結果。

<input id="text"></input>
<script>
    var text = document.querySelector('#text');
    text.addEventListener('keyup', (e) =>{
        var
searchText = e.target.value;
// 傳送輸入內容到後臺 $.ajax({ url: `search.qq.com/${searchText}`, success: data => { // 拿到後臺返回資料,並展示搜尋結果 render(data); } }); }); </script>

上面程式碼實現我們要的功能,但存在兩個較大的問題:

  1. 多餘的請求
    當想搜尋“愛迪生”時,輸入框可能會存在三種情況,“愛”、“愛迪”、“愛迪生”。而這三種情況將會發起 3 次請求,存在 2 次多餘的請求。

  2. 已無用的請求仍然執行
    一開始搜了“愛迪生”,然後馬上改搜尋“達爾文”。結果後臺返回了“愛迪生”的搜尋結果,執行渲染邏輯後結果框展示了“愛迪生”的結果,而不是當前正在搜尋的“達爾文”,這是不正確的。

減少多餘請求數,可以用 setTimeout 函式節流的方式來處理,核心程式碼如下

<input id="text"></input>
<script>
    var text = document.querySelector('#text'),
        timer = null;
    text.addEventListener('keyup', (e) =>{
        // 在 250 毫秒內進行其他輸入,則清除上一個定時器
        clearTimeout(timer);
        // 定時器,在 250 毫秒後觸發
        timer = setTimeout(() => {
            console.log('發起請求..');
        },250)
    })
</script>

已無用的請求仍然執行的解決方式,可以在發起請求前宣告一個當前搜尋的狀態變數,後臺將搜尋的內容及結果一起返回,前端判斷返回資料與當前搜尋是否一致,一致才走到渲染邏輯。最終程式碼為

<input id="text"></input>
<script>
    var text = document.querySelector('#text'),
        timer = null,
        currentSearch = '';

    text.addEventListener('keyup', (e) =>{
        clearTimeout(timer)
        timer = setTimeout(() => {
            // 宣告一個當前所搜的狀態變數
            currentSearch = ''; 

            var searchText = e.target.value;
            $.ajax({
                url: `search.qq.com/${searchText}`,
                success: data => {
                    // 判斷後臺返回的標誌與我們存的當前搜尋變數是否一致
                    if (data.search === currentSearch) {
                        // 渲染展示
                        render(data);
                    } else {
                        // ..
                    }
                }           
            });
        },250)
    })
</script>

上面程式碼基本滿足需求,但程式碼開始顯得亂糟糟。我們來使用 RxJS 實現上面程式碼功能,如下

var text = document.querySelector('#text');
var inputStream = Rx.Observable.fromEvent(text, 'keyup')
                    .debounce(250)
                    .pluck('target', 'value')
                    .flatMapLatest(url => Http.get(url))
                    .subscribe(data => render(data));

可以明顯看出,基於 RxJS 的實現,程式碼十分簡潔!

RxJS · 流 Stream

RxJS 是 Reactive Extensions for JavaScript 的縮寫,起源於 Reactive Extensions,是一個基於可觀測資料流在非同步程式設計應用中的庫。RxJS 是 Reactive Extensions 在 JavaScript 上的實現,而其他語言也有相應的實現,如 RxJava、RxAndroid、RxSwift 等。學習 RxJS,我們需要從可觀測資料流(Streams)說起,它是 Rx 中一個重要的資料型別。

是在時間流逝的過程中產生的一系列事件。它具有時間與事件響應的概念。

rxjs_stream

下雨天時,雨滴隨時間推移逐漸產生,下落時對水面產生了水波紋的影響,這跟 Rx 中的流是很類似的。而在 Web 中,雨滴可能就是一系列的滑鼠點選、鍵盤點選產生的事件或資料集合等等。

RxJS 基礎實現原理簡析

對流的概念有一定理解後,我們來講講 RxJS 是怎麼圍繞著流的概念來實現的,講講 RxJS 的基礎實現原理。RxJS 是基於觀察者模式和迭代器模式以函數語言程式設計思維來實現的。

觀察者模式

觀察者模式在 Web 中最常見的應該是 DOM 事件的監聽和觸發。

  • 訂閱:通過 addEventListener 訂閱 document.body 的 click 事件。
  • 釋出:當 body 節點被點選時,body 節點便會向訂閱者釋出這個訊息。
document.body.addEventListener('click', function listener(e) {
    console.log(e);
},false);

document.body.click(); // 模擬使用者點選

將上述例子抽象模型,並對應通用的觀察者模型

2016-11-01 9 53 52

迭代器模式

迭代器模式可以用 JavaScript 提供了 Iterable Protocol 可迭代協議來表示。Iterable Protocol 不是具體的變數型別,而是一種可實現協議。JavaScript 中像 Array、Set 等都屬於內建的可迭代型別,可以通過 iterator 方法來獲取一個迭代物件,呼叫迭代物件的 next 方法將獲取一個元素物件,如下示例。

var iterable = [1, 2];

var iterator = iterable[Symbol.iterator]();

iterator.next(); // => { value: "1", done: false}
iterator.next(); // => { value: "2", done: false}

iterator.next(); // => { value: undefined, done: true}

元素物件中:value 表示返回值,done 表示是否已經到達最後。

遍歷迭代器可以使用下面做法。

var iterable = [1, 2];
var iterator = iterable[Symbol.iterator]();

var iterator = iterable();

while(true) {
    try {
        let result = iterator.next();  // <= 獲取下一個值
    } catch (err) {
        handleError(err);  // <= 錯誤處理
    }
    if (result.done) {
        handleCompleted();  // <= 無更多值(已完成)
        break;
    }
    doSomething(result.value);
}

主要對應三種情況:

  • 獲取下一個值
    呼叫 next 可以將元素一個個地返回,這樣就支援了返回多次值。

  • 無更多值(已完成)
    當無更多值時,next 返回元素中 done 為 true。

  • 錯誤處理
    當 next 方法執行時報錯,則會丟擲 error 事件,所以可以用 try catch 包裹 next 方法處理可能出現的錯誤。

RxJS 的觀察者 + 迭代器模式

RxJS 中含有兩個基本概念:Observables 與 Observer。Observables 作為被觀察者,是一個值或事件的流集合;而 Observer 則作為觀察者,根據 Observables 進行處理。
Observables 與 Observer 之間的訂閱釋出關係(觀察者模式) 如下:

  • 訂閱:Observer 通過 Observable 提供的 subscribe() 方法訂閱 Observable。
  • 釋出:Observable 通過回撥 onNext 方法向 Observer 釋出事件。

下面為 Observable 與 Observer 的虛擬碼

// Observer
var Observer = {
    next(value) {
        alert(`收到${value}`);
    }
};

// Observable
function Observable (Observer) {
    setTimeout(()=>{
        Observer.next('A');
    },1000)
}

// subscribe
Observable(Observer);

上面實際也是觀察者模式的表現,那麼迭代器模式在 RxJS 中如何體現呢?

在 RxJS 中,Observer 除了有 next 方法來接收 Observable 的事件外,還可以提供了另外的兩個方法:error() 和 complete(),與迭代器模式一一對應。

var Observer = {
    next(value) { /* 處理值*/ },
    error(error) { /* 處理異常 */ },
    complete() { /* 處理已完成態 */ }
};

結合迭代器 Iterator 進行理解:

  • next()
    Observer 提供一個 next 方法來接收 Observable 流,是一種 push 形式;而 Iterator 是通過呼叫 iterator.next() 來拿到值,是一種 pull 的形式。

  • complete()
    當不再有新的值發出時,將觸發 Observer 的 complete 方法;而在 Iterator 中,則需要在 next 的返回結果中,當返回元素 done 為 true 時,則表示 complete。

  • error()
    當在處理事件中出現異常報錯時,Observer 提供 error 方法來接收錯誤進行統一處理;Iterator 則需要進行 try catch 包裹來處理可能出現的錯誤。

下面是 Observable 與 Observer 實現觀察者 + 迭代器模式的虛擬碼,資料的逐漸傳遞傳遞與影響其實就是流的表現。

// Observer
var Observer = {
    next(value) {
        alert(`收到${value}`);
    },
    error(error) {
        alert(`收到${value}`);
    },
    complete() {
        alert("complete");
    },
};

// Observable
function Observable (Observer) {
    [1,2,3].map(item=>{
        Observer.next(item);
    });

    Observer.complete();
    // Observer.error("error message");
}

// subscribe
Observable(Observer);

RxJS 基礎實現

有了上面的概念及虛擬碼,那麼在 RxJS 中是怎麼建立 Observable 與 Observer 的呢?

建立 Observable

RxJS 提供 create 的方法來自定義建立一個 Observable,可以使用 onNext 來發出流。

var Observable = Rx.Observable.create(observer => {
    observer.onNext(2);
    observer.onCompleted();
    return  () => console.log('disposed');
});

建立 Observer

Observer 可以宣告 next、err、complete 方法來處理流的不同狀態。

var Observer = Rx.Observer.create(
    x => console.log('Next:', x),
    err => console.log('Error:', err),
    () => console.log('Completed')
);

最後將 Observable 與 Observer 通過 subscribe 訂閱結合起來。

var subscription = Observable.subscribe(Observer);

RxJS 中流是可以被取消的,呼叫 subscribe 將返回一個 subscription,可以通過呼叫 subscription.unsubscribe() 將流進行取消,讓流不再產生。

看了起來挺複雜的?換一個實現形式:

// @Observables 建立一個 Observables
var streamA = Rx.Observable.just(2);

// @Observer streamA$.subscribe(Observer)
streamA.subscribe(v => console.log(v));

將上面程式碼改用鏈式寫法,程式碼變得十分簡潔:

Rx.Observable.just(2).subscribe(v => console.log(v));

RxJS · Operators 操作

Operators 操作·入門

Rx.Observable.just(2).subscribe(v => console.log(v));

上面程式碼相當於建立了一個流(2),最終打印出2。那麼如果想將列印結果翻倍,變成4,應該怎麼處理呢?

方案一?: 改變事件源,讓 Observable 值 X 2

Rx.Observable.just(2 * 2 /* <= */).subscribe(v => console.log(v));

方案二?: 改變響應方式,讓 Observer 處理 X 2

Rx.Observable.just(2).subscribe(v => console.log(v * 2 /* <= */));

優雅方案: RxJS 提供了優雅的處理方式,可以在事件源(Observable)與響應者(Observer)之間增加操作流的方法。

Rx.Observable.just(2)
             .map(v => v * 2) /* <= */
             .subscribe(v => console.log(v));

map 操作跟陣列操作的作用是一致的,不同的這裡是將流進行改變,然後將新的流傳出去。在 RxJS 中,把這類操作流的方式稱之為 Operators(操作)。RxJS提供了一系列 Operators,像map、reduce、filter 等等。操作流將產生新流,從而保持流的不可變性,這也是 RxJS 中函數語言程式設計的一點體現。關於函數語言程式設計,這裡暫不多講,可以看看另外一篇文章 《談談函數語言程式設計》

到這裡,我們知道了,流從產生到最終處理,可能經過的一些操作。即 RxJS 中 Observable 將經過一系列 Operators 操作後,到達 Observer。

          Operator1   Operator2
Observable ----|-----------|-------> Observer

一系列的 Operators 操作

RxJS 提供了非常多的操作,像下面這些。

Aggregate,All,Amb,ambArray,ambWith,AssertEqual,averageFloat,averageInteger,averageLong,blocking,blockingFirst,blockingForEach,blockingSubscribe,Buffer,bufferWithCount,bufferWithTime,bufferWithTimeOrCount,byLine,cache,cacheWithInitialCapacity,case,Cast,Catch,catchError,catchException,collect,concatWith,Connect,connect_forever,cons,Contains,doAction,doAfterTerminate,doOnComplete,doOnCompleted,doOnDispose,doOnEach,doOnError,doOnLifecycle,doOnNext,doOnRequest,dropUntil,dropWhile,ElementAt,ElementAtOrDefault,emptyObservable,fromNodeCallback,fromPromise,fromPublisher,fromRunnable,Generate,generateWithAbsoluteTime,generateWithRelativeTime,Interval,intervalRange,into,latest (Rx.rb version of Switch),length,mapTo,mapWithIndex,Materialize,Max,MaxBy,mergeArray,mergeArrayDelayError,mergeWith,Min,MinBy,multicastWithSelector,nest,Never,Next,Next (BlockingObservable version),partition,product,retryWhen,Return,returnElement,returnValue,runAsync,safeSubscribe,take_with_time,takeFirst,TakeLast,takeLastBuffer,takeLastBufferWithTime,windowed,withFilter,withLatestFrom,zipIterable,zipWith,zipWithIndex

關於每一個操作的含義,可以檢視官網進行了解。下面將舉幾個例子。

just
just 可以將普通資料轉換成流式資料 Observable。如上面的 Rx.Observable.just(2)。

fromEvent
除了數值外,RxJS 還提供了關於事件的操作,fromEvent 可以用來監聽事件。當事件觸發時,將事件 event 轉成可流動的 Observable 進行傳輸。下面示例表示:監聽文字框的 keyup 事件,觸發 keyup 可以產生一系列的 event Observable。

var text = document.querySelector('#text');
Rx.Observable.fromEvent(text, 'keyup')
             .subscribe(e => console.log(e));

map
map 方法跟我們平常使用的方式是一樣的,不同的只是這裡是將流進行改變,然後將新的流傳出去。上面示例已有涉及,這裡不再多講。

Rx.Observable.just(2)
             .map(v => 10 * v)
             .subscribe(v => console.log(v));

Rx 提供了許多的操作,為了更好的理解各個操作的作用,我們可以通過一個視覺化的工具 marbles 圖 來輔助理解。如 map 方法對應的 marbles 圖如下

map

箭頭可以理解為時間軸,上面的資料經過中間的操作,轉變成下面的模樣。

FlatMap
FlatMap 也是 RxJS 中常用的介面,我們來結合 marbles 圖來理解它

rxjs_flatmap

上面的資料流中,產生了新的分支流(流中流),FlatMap 的作用則是將分支流調整回主幹上,最終分支上的資料流都經過主幹的其他操作,其實也是將流中流進行扁平化。

FlatMapLatest
FlatMapLatest 與 FlatMap 都是將分支流疏通到主幹上,而不同的地方在於 FlatMapLatest 只會保留最後的流,而取消拋棄之前的流。

除了上面提到的 marbles,也可以 ASCII 字元的方式來繪製視覺化圖表,下面將結合 Map、FlatMap 和 FlatMapLatest 進行對比來理解。

@Map             @FlatMap            @FlatMapLatest
                         ↗  ↗                 ↗  ↗
-A------B-->           a2 b2                a2 b2  
-2A-----2B->          /  /                 /  /  
                    /  /                 /  /
                  a1 b1                a1 b1
                 /  /                 /  /
                -A-B----------->     -A-B---------->
                --a1-b1-a2-b2-->     --a1-b1---b2-->

FlatMap 和 FlatMapLatest 中,A 和 B 是主幹上產生的流,a1、a2 為 A 在分支上產生,b1、b2 為 B 在分支上產生,可看到,最終將歸併到主幹上。FlatMapLatest 只保留最後的流,所以將 A 的 a2 拋棄掉。

Debounce
Debounce 操作可以操作一個時間戳 TIMES,表示經過 TIMES 毫秒後,沒有流入新值,那麼才將值轉入下一個操作。

rxjs_debounce

RxJS 中的操作符是滿足我們以前的開發思維的,像 map、reduce 這些。另外,無論是 marbles 圖還是用 ASCII 字元圖這些視覺化的方式,都對 RxJS 的學習和理解有非常大的幫助。

使用 RxJS 一步步實現搜尋示例

RxJS 提供許多建立流或操作流的介面,應用這些介面,我們來一步步將搜尋的示例進行 Rx 化。

使用 RxJS 提供的 fromEvent 介面來監聽我們輸入框的 keyup 事件,觸發 keyup 將產生 Observable。

var text = document.querySelector('#text');
Rx.Observable.fromEvent(text, 'keyup')
             .subscribe(e => console.log(e));

這裡我們並不想輸出事件,而想拿到文字輸入值,請求搜尋,最終渲染出結果。涉及到兩個新的 Operators 操作,簡單理解一下:

  • pluck('target', 'value')
    將輸入的 event,輸出成 event.target.value。

  • flatMap()
    將請求搜尋結果輸出回給 Observer 上進行渲染。

var text = document.querySelector('#text');
Rx.Observable.fromEvent(text, 'keyup')
             .pluck('target', 'value') // <--
             .flatMap(url => Http.get(url)) // <--
             .subscribe(data => render(data))

上面程式碼實現了簡單搜尋呈現,但同樣存在一開始提及的兩個問題。那麼如何減少請求數,以及取消已無用的請求呢?我們來了解 RxJS 提供的其他 Operators 操作,來解決上述問題。

  • debounce(TIMES)
    表示經過 TIMES 毫秒後,沒有流入新值,那麼才將值轉入下一個環節。這個與前面使用 setTimeout 來實現函式節流的方式有一致效果。

  • flatMapLatest()
    使用 flatMapLatest 替換 flatMap,將能取消上一個已無用的請求,只保留最後的請求結果流,這樣就確保處理展示的是最後的搜尋的結果。

最終實現如下,與一開始的實現進行對比,可以明顯看出 RxJS 讓程式碼變得十分簡潔。

var text = document.querySelector('#text');
Rx.Observable.fromEvent(text, 'keyup')
             .debounce(250) // <- throttling behaviour
             .pluck('target', 'value')
             .flatMapLatest(url => Http.get(url)) // <- Kill the previous requests
             .subscribe(data => render(data))

總結

本篇作為 RxJS 入門篇到這裡就結束,關於 RxJS 中的其他方面內容,後續再拎出來進一步分析學習。
RxJS 作為一個庫,可以與眾多框架結合使用,但並不是每一種場合都需要使用到 RxJS。複雜的資料來源