RxJS 是一個響應式的庫，它接收從事件源發出的一個個事件，經過處理管道的層層處理之后，傳入最終的接收者，這個處理管道是由操作符組成的，開發者只需要選擇和組合操作符就能完成各種異步邏輯，極大簡化了異步編程。除此以外，RxJS 的設計還遵循了函數式、流的理念。

直接講概念比較難理解，不如我們實現一個簡易的 RxJS 再來看這些。

RxJS 的使用

RxJS 會對事件源做一層封裝，叫做 Observable，由它發出一個個事件。

比如這樣：

const source = new Observable((observer) => { let i = 0; setInterval(() => { observer.next(++i); }, 1000); });

在回調函數里面設置一個定時器，不斷通過 next 傳入事件。

這些事件會被接受者監聽，叫做 Observer。

const subscription = source.subscribe({ next: (v) => console.log(v), error: (err) => console.error(err), complete: () => console.log('complete'), });

observer 可以接收 next 傳過來的事件，傳輸過程中可能有 error，也可以在這里處理 error，還可以處理傳輸完成的事件。

這樣的一個監聽或者說訂閱，叫做 Subscription。

可以訂閱當然也可以取消訂閱：

subscription.unsubscribe();

取消訂閱時的回調函數是在 Observable 里返回的：

const source = new Observable((observer) => { let i = 0; const timer = setInterval(() => { observer.next(++i); }, 1000); return function unsubscribe() { clearInterval(timer); }; });

發送事件、監聽事件只是基礎，處理事件的過程才是 RxJS 的精髓，它設計了管道的概念，可以用操作符 operator 來組裝這個管道：

source.pipe( map((i) => ++i), map((i) => i * 10) ).subscribe(() => { //... })

事件經過管道之后才會傳到 Observer，在傳輸過程中會經過一個個操作符的處理。

比如這里的處理邏輯是，對傳過來的數據加 1，然后再乘以 10。

綜上，使用 RxJS 的代碼就是這樣的：

const source = new Observable((observer) => { let i = 0; const timer = setInterval(() => { observer.next(++i); }, 1000); return function unsubscribe() { clearInterval(timer); }; }); const subscription = source.pipe( map((i) => ++i), map((i) => i * 10) ).subscribe({ next: (v) => console.log(v), error: (err) => console.error(err), complete: () => console.log('complete'), }); setTimeout(() => { subscription.unsubscribe(); }, 4500);

我們通過 Observable 創建了一個事件源，每秒發出一個事件，這些事件會經過管道的處理再傳遞給 Observer，管道的組成是兩個 map 操作符，對數據做了 + 1 和 * 10 的處理。

Observer 接收到傳遞過來的數據，做了打印，還對錯誤和結束時的事件做了處理。此外，Observable 提供了取消訂閱時的處理邏輯，當我們在 4.5s 取消訂閱時，就可以清除定時器。

使用 RxJS 基本就是這個流程，那它是怎么實現的呢?

80 行代碼實現

RxJS先從事件源開始，實現 Observable：

觀察下它的特點：

1. 它接收一個回調函數，里面可以調用 next 來傳輸數據。
2. 它有 subscribe 方法可以用來添加 Observer 的訂閱，返回 subscription
3. 它可以在回調函數里返回 unsbscribe 時的處理邏輯
4. 它有 pipe 方法可以傳入操作符

我們按照這些特點來實現下：

首先，Observable 的構造函數要接收回調函數 _subscribe，但是不是立刻調用，而是在 subscribe 的時候才調用：

class Observable { constructor(_subscribe) { this._subscribe = _subscribe; } subscribe() { this._subscribe(); } }

回調函數的參數是有 next、error、complete 方法的對象，用于傳遞事件：

class Observable { constructor(_subscribe) { this._subscribe = _subscribe; } subscribe(observer) { const subscriber = new Subscriber(observer); this._subscribe(subscriber); } } class Subscriber{ constructor(observer) { super(); this.observer = observer; this.isStopped = false; } next(value) { if (this.observer.next && !this.isStopped) { this.observer.next(value); } } error(value) { this.isStopped = true; if (this.observer.error) { this.observer.error(value); } } complete() { this.isStopped = true; if (this.observer.complete) { this.observer.complete(); } if (this.unsubscribe) { this.unsubscribe(); } } }

這樣，在回調函數里面就可以調用 next、error、complete 方法了：

此外，回調函數的返回值是 unsbscribe 時的處理邏輯，要收集起來，在取消訂閱時調用：

class Subscription { constructor() { this._teardowns = []; } unsubscribe() { this._teardowns.forEach((teardown) => { typeof teardown === 'function' ? teardown() : teardown.unsubscribe() }); } add(teardown) { if (teardown) { this._teardowns.push(teardown); } } }

提供 unsubscribe 方法用于取消訂閱，_teardowns 用于收集所有的取消訂閱時的回調，在 unsubscribe 時調用所有 teardown 回調。

這段邏輯比較通用，可以作為 Subscriber 的父類。

然后，在 Observable 里調用 add 來添加 teardown，并且返回 subscription(它有 unsubscribe 方法)：

class Observable { constructor(_subscribe) { this._subscribe = _subscribe; } subscribe(observer) { const subscriber = new Subscriber(observer); subscriber.add(this._subscribe(subscriber)); return subscriber; } } class Subscriber extends Subscription { constructor(observer) { super(); this.observer = observer; this.isStopped = false; } next(value) { if (this.observer.next && !this.isStopped) { this.observer.next(value); } } error(value) { this.isStopped = true; if (this.observer.error) { this.observer.error(value); } } complete() { this.isStopped = true; if (this.observer.complete) { this.observer.complete(); } if (this.unsubscribe) { this.unsubscribe(); } } } class Subscription { constructor() { this._teardowns = []; } unsubscribe() { this._teardowns.forEach((teardown) => { typeof teardown === 'function' ? teardown() : teardown.unsubscribe() }); } add(teardown) { if (teardown) { this._teardowns.push(teardown); } } }

這樣，我們就實現了 Observable 和 Observer，只寫了 50 行代碼。先來測試下：

const source = new Observable((observer) => { let i = 0; const timer = setInterval(() => { observer.next(++i); }, 1000); return function unsubscribe() { clearInterval(timer); }; }); const subscription = source.subscribe({ next: (v) => console.log(v), error: (err) => console.error(err), complete: () => console.log('complete'), }); setTimeout(() => { subscription.unsubscribe(); }, 4500);

Observer 監聽到了 Observable 傳遞過來的 1、2、3、4 的數據，因為在 4.5s 時取消了訂閱，所以后面就不再有數據了。

我們用 50 行實現了基礎的 RxJS!

當然，最精髓的 operator 還沒有實現，接下來繼續完善。

我們給 Observable 添加 pipe 方法，它會調用傳入的 operator，并且上個的結果是下個的輸入，這樣就串起來了，也就是管道的概念：

class Observable { constructor(_subscribe) { //... } subscribe(observer) { //... } pipe(...operations) { return pipeFromArray(operations)(this); } } function pipeFromArray(fns) { if (fns.length === 0) { return (x) => x; } if (fns.length === 1) { return fns[0]; } return (input) => { return fns.reduce((prev, fn) => fn(prev), input); }; }

當傳入的參數是 0 個的時候，就直接返回之前的 Observable，1 個的時候直接返回，否則就通過 reduce 的方式串聯起來，組成管道。

operator 的實現就是監聽上一個 Observable，返回一個新的。

比如 map 的實現，就是傳入 project 對 value 做處理，把結果用 next 傳下去：

function map(project) { return (observable) => new Observable((subscriber) => { const subcription = observable.subscribe({ next(value) { return subscriber.next(project(value)); }, error(err) { subscriber.error(err); }, complete() { subscriber.complete(); }, }); return subcription; }); }

這樣我們就實現了 operator，來測試下：

我們調用了 pipe 方法，使用兩個 map 操作符來組織處理流程，對數據做了 +1 和 *10 的處理。

所以，Observable 傳遞過來的 1、2、3、4 傳遞給 Observer 的時候就變成了 20、30、40、50。

至此，我們實現了 RxJS 的 Observable、Observer、Subscription、operator 等概念，是一個簡易版 RxJS 了。只用了 80 行代碼。

再來看最開始的那些理念：

為什么叫做響應式呢?

因為是對事件源做監聽和一系列處理的，這種編程模式就叫做響應式。

為什么叫函數式呢?

因為每一步 operator 都是純函數，返回一個新的 Observable，這符合函數式的不可變，修改后返回一個新的的理念。

為什么叫流呢?

因為一個個事件是動態產生和傳遞的，這種數據的動態產生和傳遞就可以叫做流。

完整代碼如下：

function pipeFromArray(fns) { if (fns.length === 0) { return (x) => x; } if (fns.length === 1) { return fns[0]; } return (input) => { return fns.reduce((prev, fn) => fn(prev), input); }; } class Subscription { constructor() { this._teardowns = []; } unsubscribe() { this._teardowns.forEach((teardown) => { typeof teardown === 'function' ? teardown() : teardown.unsubscribe() }); } add(teardown) { if (teardown) { this._teardowns.push(teardown); } } } class Subscriber extends Subscription { constructor(observer) { super(); this.observer = observer; this.isStopped = false; } next(value) { if (this.observer.next && !this.isStopped) { this.observer.next(value); } } error(value) { this.isStopped = true; if (this.observer.error) { this.observer.error(value); } } complete() { this.isStopped = true; if (this.observer.complete) { this.observer.complete(); } if (this.unsubscribe) { this.unsubscribe(); } } } class Observable { constructor(_subscribe) { this._subscribe = _subscribe; } subscribe(observer) { const subscriber = new Subscriber(observer); subscriber.add(this._subscribe(subscriber)); return subscriber; } pipe(...operations) { return pipeFromArray(operations)(this); } } function map(project) { return (observable) => new Observable((subscriber) => { const subcription = observable.subscribe({ next(value) { return subscriber.next(project(value)); }, error(err) { subscriber.error(err); }, complete() { subscriber.complete(); }, }); return subcription; }); } const source = new Observable((observer) => { let i = 0; const timer = setInterval(() => { observer.next(++i); }, 1000); return function unsubscribe() { clearInterval(timer); }; }); const subscription = source.pipe( map((i) => ++i), map((i) => i * 10) ).subscribe({ next: (v) => console.log(v), error: (err) => console.error(err), complete: () => console.log('complete'), }); setTimeout(() => { subscription.unsubscribe(); }, 4500);

總結

為了理解 RxJS 的響應式、函數式、流等理念，我們實現了簡易版的 RxJS。

我們實現了 Observable、Observer、Subscription 等概念，完成了事件的產生和訂閱以及取消訂閱。

接著又實現了 operator 和 pipe，每個 operator 返回一個新的 Observable，對數據做層層處理。

寫完以后，我們能更清晰的理解響應式、函數式、流等理念在 RxJS 里是怎么體現的。

實現簡易版 RxJS，只需要 80 行代碼。

原文地址:https://mp.weixin.qq.com/s/G-Af4gDuBuXuHMTV-de-0w