RxJS入門

時間 2019-11-16

標籤 rxjs 入門简体版

原文原文鏈接

RxJS是什麼？

官方：RxJS 是 Reactive Extensions for JavaScript 的縮寫，起源於 Reactive Extensions，是一個基於可觀測數據流在異步編程應用中的庫。RxJS 是 Reactive Extensions 在 JavaScript 上的實現。

通常說到RxJS，都會講他是基於流的響應式的結合觀察者和迭代器模式的一種庫。因此下面會從這幾個關鍵詞來說。javascript

前置知識點

響應式編程（RP —— Reactive Programming）

響應式編程是一種面向數據流和變化傳播的編程範式。在編程語言中很方便地表達靜態或動態的數據流，而相關的計算模型會自動將變化的值經過數據流進行傳播。—— wikipedia

響應式編程是使用異步數據流進行編程。常見的異步數據流包含Event buses。用包含這些事件在內的任何東西建立數據流（Data stream），監聽他並做出響應。
只關注業務邏輯互相依賴的事件而不是實現細節
適用於大量和數據有關的事件交互，特別是高實時性要求

流

以點擊按鈕這一個事件流舉例：

屢次點擊按鈕事件流舉例：

一個流就是一個不間斷的按照時間排序的序列。它產生三種不一樣類型事件: 值，錯誤，完成的信號。對這三個定義事件處理函數，就能夠異步的捕獲這些事件
每一個stream有多個方法，調用時會基於原來的流返回一個新的流，原來的流不作修改，保證不可變性
數據流支持鏈式調用，你能夠組合不一樣的函數來處理流，建立和過濾不一樣的流。甚至一個流或多個流能夠做爲另一個流的輸入。你能夠合併兩個數據流。你能夠過濾一個數據流，從中獲取一個包含你感興趣的事件的數據流。你能夠未來自一個數據流的值映射到另一個數據流

觀察者模式（發佈-訂閱模式）

例子：購房者和售房部之間的信息訂閱。購房者訂閱售房部的房價信息，售房部維護一張須要信息的客戶表，當有信息時，遍歷表給符合條件的購房者推送發佈房屋信息。這裏，購房者擔任觀察者的角色，售房部是被觀察的角色，當售房部信息發生變化，則自動推送信息給購房者。

結論：流(售房部/rx.js的Observable)是被觀察的，某個函數訂閱流的某個事件（推送房價），該函數是觀察者(購房者/rx.js的Observer)。當流的某個事件產生了，對應的函數就會被執行。

迭代器模式

提供一種方法順序訪問一個聚合對象中的各個元素，而不須要暴露該對象的內部表示。最多見的就是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}複製代碼

由上可知JavaScript 的 Iterator 只有一個 next 方法，這個 next 方法只會回傳這兩種結果。iterator經過調用next獲取值，是一種pull數據的形式。

與Promise的區別

Promise本質上也是一個Observable，能使用fromPromise把Promise轉成Observable
可是Promise .then()只能返回一個值，Observable能夠返回多個值
Promise要麼resolve要麼reject，而且只響應一次。而Observable能夠響應屢次
Promise不能取消，Observable能夠調用unsubscribe()取消訂閱

解決的問題

同步和異步的統一
可組合的數據變動過程
數據和視圖的精確綁定
條件變動以後的自動從新計算

核心概念

概述關係

Demo：一個簡單的例子

Observable —— 被觀察者

Rxjs是觀察者 + 迭代器模式的結合，Observable做爲被觀察者，是一個值或事件的流集合。就像是一個序列，裡面的元素會隨着時間推送。

var observable = Rx.Observable
// 經過create方法建立一個Observable
// 回調函數會接受observer參數，也就是觀察者角色
	.create(function(observer) {
		observer.next('hi');
		observer.next('world');

        setTimeout(() => {
			observer.next('這一段是異步操做');
		}, 30)
	})

// 訂閱這個 observable
// 只有在訂閱以後，纔會在流Observable變化的時候，調用observer提供的方法，並通知他 
// 訂閱以後也能夠取消訂閱，調用unsubscribe()便可
console.log('start')
var subscription = observable.subscribe(function(value) {
	console.log(value);
})
console.log('end')
setTimeOut(()=> {
  subscription.unsubscribe()
}, 5000)


// 程序會依次輸出
'start'
"hi"
'world'
'end'
'這一段是異步操做'複製代碼

因此，Observable不一樣於觀察者模式中的被觀察者，他沒有一份須要維護訂閱者的清單，他只是一個函數。想要訂閱他只須要傳進回調函數observer就好。而且，Observable 能夠同時處理同步和異步操做！

同時，有不少建立Observable的方法，經常使用的以下：

Operator —— 操做符

操做Observable的函數就是操做符。他會接受傳入的Observable，可是會返回新的Observable。用map舉例說明。

Rx.Observable.of(2)
             .map(v => v * 2) 
             .subscribe(v => console.log('output:' + v));
// output:4複製代碼

下面介紹幾個經常使用的操做符。具體的API可視化數據流動可參見寶石圖

Observer —— 觀察者

和迭代器模式一一對應，提供三個方法，next、error、complete

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

next(): 接收Observable發出的值  （必傳）
error(): 不一樣於迭代器裏面用try catch，Observer用error方法接收錯誤 （可選）
complete(): 當沒有新的數據發出的時候，觸發操做  （可選）複製代碼

Subject —— 觀察模式的實現，並繼承Observable

同一個Observable能夠被多個observer訂閱。和addListener相似她們由Subject維護列表，Subject能夠向多個Observer多路推送數值，是一類特殊的Observable。

每個Subject都是一個Observable，能夠訂閱他。從Observer的視角看，它並不能區分本身的執行環境是普通Observable的單路推送仍是基於Subject的多路推送
每個Subject也能夠是Observer，由於他一樣由next、error、complete方法組成，調用next方法，Subject會給全部在他上面註冊的Observer多路推送當前的值

demo地址

// 建立一個Observable，一秒鐘輸出一個數字，只取三個就結束
var source = Rx.Observable.interval(1000).take(3);

// 定義兩個observer對象
var observerA = {
    next: value => console.log('A next: ' + value),
    error: error => console.log('A error: ' + error),
    complete: () => console.log('A complete!')
}

var observerB = {
    next: value => console.log('B next: ' + value),
    error: error => console.log('B error: ' + error),
    complete: () => console.log('B complete!')
}

// 建立一個subject —— 特殊的Observable
var subject = new Rx.Subject()

// observerA訂閱Subject
subject.subscribe(observerA)

// Subject又以observer的身份訂閱Observable
source.subscribe(subject);

setTimeout(() => {
    subject.subscribe(observerB);
}, 1000);

// 輸出：
// "A next: 0"
// "A next: 1"
// "B next: 1"
// "A next: 2"
// "B next: 2"
// "A complete!"
// "B complete!"
 A、B兩個observer互不影響，是獨立的複製代碼

Scheduler —— 控制Observable的時間節點

控制一個Observable的訂閱開始執行和值送達的時機。RxJS 5中提供了4種scheduler。通常operator會有預設的scheduler。

queue —— 遞歸的時候，queue會把遞歸阻塞，避免沒必要要的性能損耗
asap —— as soon as possible，表現爲setTimeout時間爲0。多用於永不退訂的Observable，好比輪詢
async —— 規定當前的Observable執行方式爲異步，用setInterval實現。
animationFrame —— Window.requestAnimationFrame這個API實現，適合高頻率的UI動畫觸發

var observable = Rx.Observable.create(function (observer) {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();
});

console.log('before subscribe');
observable.observeOn(Rx.Scheduler.async) // 原本是同步的，變成了異步
.subscribe({
    next: (value) => { console.log(value); },
    error: (err) => { console.log('Error: ' + err); },
    complete: () => { console.log('complete'); }
});
console.log('after subscribe');

// "before subscribe"
// "after subscribe"
// 1
// 2
// 3
// "complete"複製代碼