RxJS速成 (上)

What is RxJS?

RxJS是ReactiveX編程理念的JavaScript版本。ReactiveX是一種針對異步數據流的編程。簡單來說，它將一切數據，包括HTTP請求，DOM事件或者普通數據等包裝成流的形式，然后用強大豐富的操作符對流進行處理，使你能以同步編程的方式處理異步數據，並組合不同的操作符來輕松優雅的實現你所需要的功能

下面廢話不說, 直接切入正題.

 

准備項目

我使用typescript來介紹rxjs. 因為我主要是在angular項目里面用ts.

全局安裝typescript:

npm install -g typescript

全局安裝ts-node:

npm install -g ts-node

建立一個文件夾learn-rxjs, 進入並執行:

npm init

安裝rxjs:

npm install rxjs --save

 

RxJS的主要成員

• Observable: 一系列值的生產者
• Observer: 它是observable值的消費者
• Subscriber: 連接observer和observable
• Operator: 可以在數據流的途中對值進行轉換的操作符
• Subject: 既包括Observable也包括Observer

Observable, Observer, Subscriber的角色關系:

工廠生產雜志, 郵遞員去送雜志, 就相當於是Observable, 郵遞員給你帶來了啥? 帶來了雜志, 然后(next)雜志, next雜志.....

把雜志帶給了誰? 看看這對夫婦, 可能是丈夫來付賬單訂雜志, 他就是Subscriber. 而這本女性雜志肯定不是丈夫來看(如果他是正經丈夫的話), 而妻子沒有直接去訂閱雜志, 但是她看這本雜志有用(知道怎么去用它).

所以可以這樣理解, 丈夫(Subscriber)把Observable和Observer聯系到了一起, 就是Subscriber為Observable提供了一個Observer(丈夫訂雜志, 告訴快遞員把貨給他媳婦就行).

Observable可以在Observer上調用三種方法(快遞員跟他妻子可能會有三種情況...好像這么說不太恰當), 當Observable把數據(雜志)傳遞過來的時候, 這三種情況是:

• next(), 這期雜志送完了, 等待下一期吧
• error(), 送雜志的時候出現問題了, 沒送到.
• complete(), 訂的雜志都處理完了, 以后不送了.

下面這個圖講的就是從Observable訂閱消息, 並且在Observer里面處理它們:

Observable允許:

• 訂閱/取消訂閱它的數據流
• 發送下一個值給Observer
• 告訴Observer發生了錯誤以及錯誤的信息
• 告訴Observer整個流結束了.

Observer可以提供:

• 一個可以處理流(stream)上的next的值的function
• 處理錯誤的function
• 處理流結束的function

 

創建Observable

• Observable.from(), 把數組或iterable對象轉換成Observable
• Observable.create(), 返回一個可以在Observer上調用方法的Observable.
• Observable.fromEvent(), 把event轉換成Observable.
• Observable.fromPromise(), 把Promise轉換成Observable.
• Observable.range(), 在指定范圍內返回一串數.

Observable.from()

observable_from.ts:

import { Observable } from "rxjs/Observable"; // 這里沒有使用Rx對象而是直接使用其下面的Observable對象, 因為Rx里面很多的功能都用不上.
import 'rxjs/add/observable/from'; // 這里我需要使用from 操縱符(operator)

let persons = [
    { name: 'Dave', age: 34, salary: 2000 },
    { name: 'Nick', age: 37, salary: 32000 },
    { name: 'Howie', age: 40, salary: 26000 },
    { name: 'Brian', age: 40, salary: 30000 },
    { name: 'Kevin', age: 47, salary: 24000 },
];

let index = 1;
Observable.from(persons)
    .subscribe( person => { console.log(index++, person); }, err => console.log(err), () => console.log("Streaming is over.") );

subscribe里面有3個function, 這3個function就是Observer.

第一個function是指當前這個person到來的時候需要做什么;

第二個是錯誤發生的時候做什么;

第三個function就是流都走完的時候做什么.

注意, 是當執行到.subscribe()的時候, Observable才開始推送數據.

運行這個例子需要執行下面的命令:

ts-node observable_from.ts

 

Observable.create()

Observable.create是Observable構造函數的一個別名而已. 它只有一個參數就是subscribe function. 

observable_creates.ts:

import { Observable } from "rxjs/Observable";

function getData() {

    let persons = [
        { name: 'Dave', age: 34, salary: 2000 },
        { name: 'Nick', age: 37, salary: 32000 },
        { name: 'Howie', age: 40, salary: 26000 },
        { name: 'Brian', age: 40, salary: 30000 },
        { name: 'Kevin', age: 47, salary: 24000 },
    ];

    return Observable.create(
        observer => { // 這部分就是subscribe function
            persons.forEach(p => observer.next(p)); observer.complete(); }
    );
}

getData()
    .subscribe(
        person => console.log(person.name),
        err => console.error(err),
        () => console.log("Streaming is over.")
    );

create里面的部分是subscribe function. 這部分可以理解為, 每當有人訂閱這個Observable的時候, Observable會為他提供一個Observer.

在這里面, observer使用next方法對person進行推送. 當循環結束的時候, 使用complete()方法通知Observable流結束了.

盡管getDate里面create了Observable, 但是整個數據流動並不是在這時就開始的. 在這個地方, 這只不過是個聲明而已.

只有當有人去訂閱這個Observable的時候, 整個數據流才會流動.

運行該文件:

 

RxJS Operator(操作符)

 

Operator是一個function, 它有一個輸入, 還有一個輸出. 這個function輸入是Observable輸出也是Observable.

在function里面, 可以做一些轉換的動作

下面是幾個例子:

observablePersons.filter(p => p.age > 40);

這個filter function和數組的filter類似, 它接受另一個function(也可以叫predicate)作為參數, 這個function提供了某種標准, 通過這個標准可以判定是否當前的元素可以被送到訂閱者那里.

p => p.age > 40

 

這個function, 是pure function, 在functional programming(函數式編程)里面, pure function是這樣定義的: 如果參數是一樣的, 無論外界環境怎么變化, 它的結果肯定是一樣的.

pure function不與外界打交道, 不保存到數據庫, 不會存儲文件, 不依賴於時間....

而這個filter function呢, 在函數式編程里面是一個high order function.

什么是High order function? 

如果一個function的參數可以是另一個function, 或者它可以返回另一個function, 那么它就是High Order function.

 

Marble 圖

首先記住這個網址: http://rxmarbles.com/

有時候您可以通過文檔查看operator的功能, 有時候文檔不是很好理解, 這時你可以參考一下marble 圖.

例如 map:

可以看到map接受一個function作為參數, 通過該function可以把每個元素按照function的邏輯進行轉換.

 

例如 filter:

 

filter就是按條件過濾, 只讓合格的元素通過.

 

例 debounceTime (恢復時間):

如果該元素后10毫秒內, 沒有出現其它元素, 那么該元素就可以通過.

 

例 reduce:

這個也和數組的reduce是一個意思.

 

例子

import { Observable } from "rxjs/Observable"; import 'rxjs/add/observable/from'; import 'rxjs/add/operator/map'; import 'rxjs/add/operator/reduce';


let persons = [
    { name: 'Dave', age: 34, salary: 2000 },
    { name: 'Nick', age: 37, salary: 32000 },
    { name: 'Howie', age: 40, salary: 26000 },
    { name: 'Brian', age: 40, salary: 30000 },
    { name: 'Kevin', age: 47, salary: 24000 },
];

Observable.from(persons)
    .map(p => p.salary)
    .reduce((total, current) => total+ current, 0)
    .subscribe(
        totalSalary => console.log(`Total salary is ${totalSalary}`),
        err => console.log(err)
    );

 

這個例子非常的簡單, 典型的map-reduce, 就不講了.

結果如下:

 

用現實世界中煉鋼生產流程的例子來解釋使用Operator來進行Reactive數據流處理的過程:

原料(礦石)整個過程中會經過很多個工作站, 這里每個工作站都可以看作是RxJS的operator, 原料經過這些operator之后, 成品就被生產了出來.

每個工作站(operator)都是可以被組合使用的, 所以可以再加幾個工作站也行.

 

錯誤處理

Observable是會發生錯誤的, 如果錯誤被發送到了Observer的話, 整個流就結束了.

但是做Reactive編程的話, 有一個原則: Reactive的程序應該很有彈性/韌性.

也就是說, 即使錯誤發生了, 程序也應該繼續運行.

但是如果error function在Observer被調用了的話, 那就太晚了, 這樣流就停止了.

那么如何在error到達Observer之前對其進行攔截, 以便流可以繼續走下去或者說這個流停止了,然后另外一個流替它繼續走下去?

錯誤處理的Operators:

• error() 被Observable在Observer上調用
• catch() 在subscriber里並且在oserver得到它(錯誤)之前攔截錯誤,
• retry(n) 立即重試最多n次
• retryWhen(fn) 按照參數function的預定邏輯進行重試

使用catch()進行錯誤處理:

observable_catch.ts:

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

function getFromGoogle(): Observable<any> {
    return Observable.create(function subscribe(observer) {
        observer.next('https://google.com');
        observer.error({
            message: 'Google can\'t be reached.',
            status: 404,
        });
        observer.complete();
    });
}

function getFromBing(): Observable<any> {
    return Observable.create(function subscribe(observer) {
        observer.next('https://global.bing.com');
        observer.complete();
    });
}


function getFromBaidu(): Observable<any> {
    return Observable.create(function subscribe(observer) {
        observer.next('https://www.baidu.com');
        observer.complete();
    });
}

getFromGoogle()
    .catch(err => {
        console.error(`Error: ${err.status}: ${err.message}`);
        if(err.status === 404) {
            return getFromBaidu();
        } else {
            return getFromBing();
        }
    })
    .map(x => `The site is : ${x}`)
    .subscribe(
        x => console.log('Subscriber got: ' + x),
        err => console.error(err),
        () => console.log('The stream is over.')
    );

 

在subscribe之前調用catch, catch里可以進行流的替換動作.

運行結果如下:

相當於:

 

Hot 和 Cold Observable

Cold: Observable可以為每個Subscriber創建新的數據生產者

Hot: 每個Subscriber從訂閱的時候開始在同一個數據生產者那里共享其余的數據.

從原理來說是這樣的: Cold內部會創建一個新的數據生產者, 而Hot則會一直使用外部的數據生產者.

舉個例子:

Cold: 就相當於我在騰訊視頻買體育視頻會員, 可以從頭看里面的足球比賽.

Hot: 就相當於看足球比賽的現場直播, 如果來晚了, 那么前面就看不到了.

 

Share Operator

share() 操作符允許多個訂閱者共享同一個Observable. 也就是把Cold變成Hot.

例子 observable_share.ts:

import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/interval';
import 'rxjs/add/operator/take';
import 'rxjs/add/operator/share';

const numbers = Observable
    .interval(1000)
    .take(5)
    .share();

function subscribeToNumbers(name) {
    numbers.subscribe(
        x => console.log(`${name}: ${x}`)
    );
}

subscribeToNumbers('Dave');

const anotherSubscription = () => subscribeToNumbers('Nick');

setTimeout(anotherSubscription, 2500);

 

這里interval是每隔1秒產生一個數據, take(5)表示取5個數, 也就是1,2,3,4,5.

然后share()就把這個observable從cold變成了hot的.

后邊Dave進行了訂閱.

2.5秒以后, Nick進行了訂閱.

最后結果是: