rxjs 原始碼分析1-(fromEvent)

摘要： Rxjs 是使用 Observables 的響應式程式設計的庫，它使編寫非同步或基於回撥的程式碼更容易。我們現在針對 Rxjs 6 來進行原始碼分析，分析其實現的基本原理， 我們可以根據中文文件來學習 Rxjs 的基本使用，但是這個文件是 Rxjs 5 的版本。其最基本的使用區別如下,Rxj...

ReactiveX%2FRxJS" rel="nofollow,noindex">Rxjs 是使用 Observables 的響應式程式設計的庫，它使編寫非同步或基於回撥的程式碼更容易。我們現在針對 Rxjs 6 來進行原始碼分析，分析其實現的基本原理， 我們可以根據中文文件來學習 Rxjs 的基本使用，但是這個文件是 Rxjs 5 的版本。其最基本的使用區別如下,Rxjs 6的操作符都放在 pipe (管道)中配置，而Rxjs 5 的版本是直接呼叫

Rxjs 5

fromEvent(addBtn, 'click')
.throttleTime(3000)
.subscribe(() => {
nameInput.value = +(nameInput.value) + 1
})
複製程式碼

Rxjs 6

fromEvent(addBtn, 'click')
.pipe(throttleTime(3000))
.subscribe(() => {
nameInput.value = +(nameInput.value) + 1
})
複製程式碼

fromEvent

使用方法

我們先從我們最常用的事件處理開始，Rxjs 對事件的處理有一個 fromEvent 方法， 其最簡單的一個範例如下：

import {fromEvent } from './esm2015';

const addBtn = document.getElementById('add')
const minusBtn = document.getElementById('minus')
const nameInput = document.getElementById('name');
fromEvent(addBtn, 'click')
.subscribe(() => {
nameInput.value = +(nameInput.value) + 1
})

fromEvent(minusBtn, 'click')
.subscribe(() => {
nameInput.value = +(nameInput.value) - 1
})
複製程式碼

其實現的效果如下圖：

我們點選加號或者減號 button 去給Input賦值。

原理

從上面我們已經可以知道怎麼去簡單使用 fromEvent , 下面我們根據原始碼來一步步深入分析，其基本原理：

export function fromEvent(target, eventName, options, resultSelector) {
if (isFunction(options)) {
resultSelector = options;
options = undefined;
}
if (resultSelector) {
return fromEvent(target, eventName, options).pipe(map(args => isArray(args) ? resultSelector(...args) : resultSelector(args)));
}
return new Observable(subscriber => {
function handler(e) {
if (arguments.length > 1) {
subscriber.next(Array.prototype.slice.call(arguments));
}
else {
subscriber.next(e);
}
}
setupSubscription(target, eventName, handler, subscriber, options);
});
}
複製程式碼

fromEvent 其實就是一個方法， 可以傳入四個引數，我們上面的Demo 只是傳遞了兩個引數，我們先只分析只傳遞兩個引數的情況： fromEvent 最終會返回一個 Observable 物件， 我們可以將上面的Demo 程式碼，進行簡單的處理，如下：

import { fromEvent } from './esm2015';

const addBtn = document.getElementById('add')
const minusBtn = document.getElementById('minus')
const nameInput = document.getElementById('name')
const addFromEventObj = fromEvent(addBtn, 'click')
addFromEventObj.subscribe(() => {
nameInput.value = +(nameInput.value) + 1
})

const minusFromEventObj = fromEvent(minusBtn, 'click')
minusFromEventObj.subscribe(() => {
nameInput.value = +(nameInput.value) - 1
})
複製程式碼

我們將如下程式碼分成了兩個步驟，

fromEvent(addBtn, 'click')
.subscribe(() => {
nameInput.value = +(nameInput.value) + 1
})

複製程式碼

const addFromEventObj = fromEvent(addBtn, 'click') 我們可以檢視 addFromEventObj 物件，如下截圖：

很簡單，就是 Observable 物件，其中有一個重要的屬性 _subscribe 的屬性， 執行的就是 fromEvent 中 return

物件的傳入的引數：

return new Observable(subscriber => {
function handler(e) {
if (arguments.length > 1) {
subscriber.next(Array.prototype.slice.call(arguments));
}
else {
subscriber.next(e);
}
}
setupSubscription(target, eventName, handler, subscriber, options);
});
複製程式碼

其實 fromEvent 方法，很簡單，就是返回一個 Observable 物件， 其他的基本就沒有什麼了。 其中點還是返回的這個物件，我們下面來深入分析 Observable 物件.

Observable

Observable 的中文翻譯就是 可觀察的 , 表示一個可觀察的物件，既然是一個可觀察的物件，那觀察到變化後，是不是要通知相應的 觀察者 呢？

建構函式

我們首先分析 Observable 的建構函式：

constructor(subscribe) {
this._isScalar = false;
if (subscribe) {
this._subscribe = subscribe;
}
}
複製程式碼

其建構函式就接收一個引數 subscribe ，其中文意思就是 訂閱 ，其實 訂閱 就是 觀察者 的概念，然後我們的 可觀察的物件 和 觀察者 物件就關聯起來了。

const addFromEventObj = fromEvent(addBtn, 'click') 中 addFromEventObj 表示建立了一個 可觀察的 物件， 有了 可觀察的 物件，我們就需要對這個物件進行 訂閱 了， 我們下面就來分析 subscribe 方法

subscribe

subscribe的原始碼如下：

subscribe(observerOrNext, error, complete) {
const { operator } = this;
const sink = toSubscriber(observerOrNext, error, complete);
if (operator) {
operator.call(sink, this.source);
}
else {
sink.add(
this.source || 
(config.useDeprecatedSynchronousErrorHandling && !sink.syncErrorThrowable) ?
this._subscribe(sink) :
this._trySubscribe(sink)
);
}
if (config.useDeprecatedSynchronousErrorHandling) {
if (sink.syncErrorThrowable) {
sink.syncErrorThrowable = false;
if (sink.syncErrorThrown) {
throw sink.syncErrorValue;
}
}
}
return sink;
}
複製程式碼

我們目前先不考慮新增 operator (操作符)的情況，我們下面來一一分析這個方法：

1. const sink = toSubscriber(observerOrNext, error, complete); 建立了一個 Subscriber 方法，
2. 因為 operator 為undefined, 所以條件判斷語句會進入 else ,

sink.add(
this.source || 
(config.useDeprecatedSynchronousErrorHandling && !sink.syncErrorThrowable) ?
this._subscribe(sink) :
this._trySubscribe(sink)
);
複製程式碼

其中add()方法裡面會先執行 this._trySubscribe(sink) , 然後執行： return this._subscribe(sink); , 我們應該還記得 this._subscribe ,是我們呼叫 fromEvent 方法， 返回 Observable 物件，傳入建構函式的引數，程式碼如下：

subscriber => {
function handler(e) {
if (arguments.length > 1) {
subscriber.next(Array.prototype.slice.call(arguments));
}
else {
subscriber.next(e);
}
}
setupSubscription(target, eventName, handler, subscriber, options);
}
複製程式碼

這裡就是給真實的 Dom 元素進行事件的繫結，現在我們點選 button ， 就會呼叫如上的 handler 方法， 主要實現是 subscriber.next(e); , 其'subscriber' 也就是 const sink = toSubscriber(observerOrNext, error, complete); 建立的 sink 物件，其對應就是 esm2015\internal\Subscriber.js 物件，現在我們檢視其對應的 next() 方法， 其指向的是：

_next(value) {
this.destination.next(value);
}
複製程式碼

其 this.destination 屬性， 是在 Subscriber 建構函式賦值的, 是一個 SafeSubscriber 物件

default:
this.syncErrorThrowable = true;
this.destination = new SafeSubscriber(this, destinationOrNext, error, complete);
break;
複製程式碼

其中 destinationOrNext 對應的是 subscribe 方法傳入的引數，如下：

const addFromEventObj = fromEvent(addBtn, 'click')
addFromEventObj.subscribe(() => {
nameInput.value = +(nameInput.value) + 1
})
複製程式碼

就是

() => {
nameInput.value = +(nameInput.value) + 1
}
複製程式碼

下面我們繼續來分析 this.destination.next(value); , 其程式碼如下：

next(value) {
if (!this.isStopped && this._next) {
const { _parentSubscriber } = this;
if (!config.useDeprecatedSynchronousErrorHandling || !_parentSubscriber.syncErrorThrowable) {
this.__tryOrUnsub(this._next, value);
}
else if (this.__tryOrSetError(_parentSubscriber, this._next, value)) {
this.unsubscribe();
}
}
}
複製程式碼

最終實現方法是 this.__tryOrUnsub(this._next, value); , 這個方法傳遞了兩個引數：

1. this._next ， 其在 SafeSubscriber 建構函式中對其進行了賦值： next = observerOrNext; ,也就是建構函式的第二個引數， 我們上面已經分析了建立 SafeSubscriber 物件的地方 this.destination = new SafeSubscriber(this, destinationOrNext, error, complete); , 這個 destinationOrNext 就是 subscribe 方法傳入的引數，如下：

const addFromEventObj = fromEvent(addBtn, 'click')
addFromEventObj.subscribe(() => {
nameInput.value = +(nameInput.value) + 1
})
複製程式碼

2. value , 也就是 click 事件的物件 MouseEvent

我們接下來具體分析**__tryOrUnsub** 方法，程式碼如下：

__tryOrUnsub(fn, value) {
try {
fn.call(this._context, value);
}
catch (err) {
this.unsubscribe();
if (config.useDeprecatedSynchronousErrorHandling) {
throw err;
}
else {
hostReportError(err);
}
}
}
複製程式碼

其主要實現就是 fn.call(this._context, value); , 就會執行了 subscribe 裡面的方法了， 也就是執行：

() => {
nameInput.value = +(nameInput.value) + 1
}
複製程式碼

pipe

上面我們已經基本理解了 fromEvent 的基本使用方法，主要分析的是 subscribe 方法，我們現在有個需求，我們要控制Button , 在 3s 時間內， 我們只能點選一次， 以防止，惡意點選按鈕. Rxjs 都是基於流來操作， Observable 物件提供了一個 pipe (管道)的方法， 在進入到 subscribe 訂閱者方法之前，所以的資料需要進行加工，異常處理， 以保證 subscribe 收到的是正確的資料。我們下面來深入分析 pipe 方法。 我們將我們的Demo 修改如下：

import { fromEvent } from './esm2015';
import { throttleTime } from './esm2015/operators'

const addBtn = document.getElementById('add')
const minusBtn = document.getElementById('minus')
const nameInput = document.getElementById('name')
const addFromEventObj = fromEvent(addBtn, 'click')
const pipeObj = addFromEventObj
.pipe(throttleTime(1000 * 3))
pipeObj.subscribe(() => {
nameInput.value = +(nameInput.value) + 1
})

const minusFromEventObj = fromEvent(minusBtn, 'click')

minusFromEventObj
.pipe(throttleTime(1000 * 3))
.subscribe(() => {
nameInput.value = +(nameInput.value) - 1
})
複製程式碼

上面我們通過 .pipe(throttleTime(1000 * 3)) 添加了 管道 ,其中傳入了一個 throttleTime (節流)操作符， 下面是 pipe 方法的程式碼：

pipe(...operations) {
if (operations.length === 0) {
return this;
}
var opts = pipeFromArray(operations);
var result = opts(this);
return result;
}
複製程式碼

從上面方法，我們可以看出pipe 方法，可以傳入多個操作符， 我們現在先只看下簡單的，只傳入一個操作符的情況， 如果只傳入一個引數 var opts = pipeFromArray(operations); 中 opts 對應的就是我們傳入的 throttleTime 函式返回的方法，也就是如下：

export function throttleTime(duration, scheduler = async, config = defaultThrottleConfig) {
return (source) => {
return source.lift(new ThrottleTimeOperator(duration, scheduler, config.leading, config.trailing));
}
}
複製程式碼

其中 source 就是上面的 this , 也就是 Observable 物件，我們下面可以繼續看下 lift 方法：

lift(operator) {
const observable = new Observable();
observable.source = this;
observable.operator = operator;
return observable;
}
複製程式碼

返回了一個新的 observable 物件，只是在添加了 operator 屬性。如下程式碼：

const addFromEventObj = fromEvent(addBtn, 'click')
const pipeObj = addFromEventObj
.pipe(throttleTime(1000 * 3))
複製程式碼

pipeObj 如下圖所示：

所以 pipe 方法就是將一個(組)操作符掛載在一個新的 observable 物件的 operator 屬性上。 我們還需要重新分析 subscribe 方法

subscribe(observerOrNext, error, complete) {
const { operator } = this;
const sink = toSubscriber(observerOrNext, error, complete);
if (operator) {
operator.call(sink, this.source);
}
else {
sink.add(
this.source || 
(config.useDeprecatedSynchronousErrorHandling && !sink.syncErrorThrowable) ?
this._subscribe(sink) :
this._trySubscribe(sink)
);
}
if (config.useDeprecatedSynchronousErrorHandling) {
if (sink.syncErrorThrowable) {
sink.syncErrorThrowable = false;
if (sink.syncErrorThrown) {
throw sink.syncErrorValue;
}
}
}
return sink;
}
複製程式碼

我們在呼叫 subscribe 方法之前，我們已經呼叫了 pipe 方法, pipe 方法返回的物件， 已經有了 operator 操作符，所以上面的邏輯分支會走 if operator.call(sink, this.source); , call 程式碼如下：

call(subscriber, source) {
return source.subscribe(new ThrottleTimeSubscriber(subscriber, this.duration, this.scheduler, this.leading, this.trailing));
}
複製程式碼

重新呼叫了 subscribe 方法， 只是傳入的第一個引數 observerOrNext 是一個ThrottleTimeSubscriber物件， 我們重新回到 subscribe 方法， const sink = toSubscriber(observerOrNext, error, complete); , 檢視下：

export function toSubscriber(nextOrObserver, error, complete) {
if (nextOrObserver) {
if (nextOrObserver instanceof Subscriber) {
return nextOrObserver;
}
if (nextOrObserver[rxSubscriberSymbol]) {
return nextOrObserver[rxSubscriberSymbol]();
}
}
if (!nextOrObserver && !error && !complete) {
return new Subscriber(emptyObserver);
}
return new Subscriber(nextOrObserver, error, complete);
}
複製程式碼

從程式碼可以分析出，如果第一個引數 nextOrObserver 是一個 Subscriber 型別， 就直接返回這個物件， 而我們的 ThrottleTimeSubscriber 是繼承與 Subscriber 物件的，所以 sink 就是一個 ThrottleTimeSubscriber 物件。 我們又重新回到 fromEvent 方法, 其中的subscriber指向的就是一個 ThrottleTimeSubscriber 物件。 subscriber.next(e); 呼叫的也是 ThrottleTimeSubscriber 物件的 next 方法。

return new Observable(subscriber => {
function handler(e) {
if (arguments.length > 1) {
subscriber.next(Array.prototype.slice.call(arguments));
}
else {
subscriber.next(e);
}
}
setupSubscription(target, eventName, handler, subscriber, options);
});
複製程式碼

後面我們來繼續分析， 這個 operator 操作符是怎麼起作用的。

throttleTime

上面我們已經分析， pipe 方法就是將一個(組)操作符掛載在一個新的 observable 物件的 operator 屬性上。現在我們來分析操作符具體是怎麼工作的，我們可以先以 throttleTime 操作符來進行分析。

上面我們已經分析了，我們在點選Button 的時候，會呼叫 handler 方法， 其中 subscriber 已經是一個 ThrottleTimeSubscriber 物件。

return new Observable(subscriber => {
function handler(e) {
if (arguments.length > 1) {
subscriber.next(Array.prototype.slice.call(arguments));
}
else {
subscriber.next(e);
}
}
setupSubscription(target, eventName, handler, subscriber, options);
});
複製程式碼

我們下面來分析： ThrottleTimeSubscriber.next(e) 方法， 其程式碼如下：

_next(value) {
if (this.throttled) {
if (this.trailing) {
this._trailingValue = value;
this._hasTrailingValue = true;
}
}
else {
this.throttled = this.scheduler.schedule(dispatchNext, this.duration, { subscriber: this })
this.add(this.throttled);
if (this.leading) {
this.destination.next(value);
}
}
}
複製程式碼

這個方法很重要

1. 首先加了一個 throttled 標記變數， 用來標記是否已經啟動了節流開關，一開始是為 undefinded 的 程式碼會進入 else 分支，
2. 然後執行 this.throttled = this.scheduler.schedule(dispatchNext, this.duration, { subscriber: this }) 給throttled賦值，下一次進來的時候， throttled 就有值了。
3. 執行 this.destination.next(value); , 這個方法，會最終呼叫 subscribe 訂閱方法中傳遞的方法，也就是

addFromEventObj = addFromEventObj.subscribe(() => {
nameInput.value = +(nameInput.value) + 1
})
複製程式碼

總結：

這個方法是實現 throttleTime 節流的關鍵點， 通過判斷 throttled 標記來判斷是否要執行 subscribe 中的方法。

下面我們來重點分析： this.throttled = this.scheduler.schedule(dispatchNext, this.duration, { subscriber: this }) .

首先： scheduler 指向的是 esm2015\internal\scheduler\AsyncAction.js 物件， 我們檢視下其 schedule 方法：

schedule(state, delay = 0) {
if (this.closed) {
return this;
}
this.state = state;
const id = this.id;
const scheduler = this.scheduler;
if (id != null) {
this.id = this.recycleAsyncId(scheduler, id, delay);
}
this.pending = true;
this.delay = delay;
this.id = this.id || this.requestAsyncId(scheduler, this.id, delay);
return this;
}
複製程式碼

其中最重要的是： this.id = this.id || this.requestAsyncId(scheduler, this.id, delay); , 這裡呼叫了一個 requestAsyncId 方法，其程式碼如下：

requestAsyncId(scheduler, id, delay = 0) {
return setInterval(scheduler.flush.bind(scheduler, this), delay);
}
複製程式碼

哈哈，這裡設定了一個定時器 setInterval , 這就是 throttleTime 生效的一個關鍵點。

總結：

setInterval 實現的功能是定時去清除 throttled 變數值，從而達到，我們在呼叫throttleTime(1000*3)這個操作符後，點選一次按鈕後，3S內不能再次點選，但是3S後，又可以點選的原理

這個定時器的第一個引數是 scheduler.flush.bind(scheduler, this) , 其對應的是： esm2015/internal/scheduler/AsyncScheduler.js 裡的flush 方法，其程式碼如下：

flush(action) {
const { actions } = this;
if (this.active) {
actions.push(action);
return;
}
let error;
this.active = true;
do {
if (error = action.execute(action.state, action.delay)) {
break;
}
} while (action = actions.shift());
this.active = false;
if (error) {
while (action = actions.shift()) {
action.unsubscribe();
}
throw error;
}
}
複製程式碼

會去遍歷所有的actions, 然後去執行 execute 方法， 傳入的action 就是對應的 AsyncAction 物件， execute 方法如下：

execute(state, delay) {
if (this.closed) {
return new Error('executing a cancelled action');
}
this.pending = false;
const error = this._execute(state, delay);
if (error) {
return error;
}
else if (this.pending === false && this.id != null) {
this.id = this.recycleAsyncId(this.scheduler, this.id, null);
}
}
複製程式碼

然後會呼叫 const error = this._execute(state, delay); , _execute 程式碼如下：

_execute(state, delay) {
let errored = false;
let errorValue = undefined;
try {
this.work(state);
}
catch (e) {
errored = true;
errorValue = !!e && e || new Error(e);
}
if (errored) {
this.unsubscribe();
return errorValue;
}
}
複製程式碼

其中最重要的是呼叫了 this.work(state) 方法， work 是 this.throttled = this.scheduler.schedule(dispatchNext, this.duration, { subscriber: this }) 傳遞的第一個引數

function dispatchNext(arg) {
const { subscriber } = arg;
subscriber.clearThrottle();
}
複製程式碼

最終會呼叫 clearThrottle 方法，其程式碼如下：

clearThrottle() {
const throttled = this.throttled;
if (throttled) {
if (this.trailing && this._hasTrailingValue) {
this.destination.next(this._trailingValue);
this._trailingValue = null;
this._hasTrailingValue = false;
}
throttled.unsubscribe();
this.remove(throttled);
this.throttled = null;
}
}
複製程式碼

其中很關鍵的一個步驟是將throttled 這個標記，設定為null this.throttled = null; , 為什麼說很重要呢， 還記得我們上面有分析 ThrottleTimeSubscriber , 這個class 下面的 _next 方法，其程式碼如下：

_next(value) {
if (this.throttled) {
if (this.trailing) {
this._trailingValue = value;
this._hasTrailingValue = true;
}
}
else {
this.throttled = this.scheduler.schedule(dispatchNext, this.duration, { subscriber: this })
this.add(this.throttled);
if (this.leading) {
this.destination.next(value);
}
}
}
複製程式碼

這個方法有判斷 throttled 這個標記， 如果不為空，其實相當與什麼也不做，只有為空的情況下，才會去執行 this.throttled = this.scheduler.schedule(dispatchNext, this.duration, { subscriber: this }) , 也就是這個方法，會去執行我們 subscribe 訂閱裡面的方法， 從而達到了，節流的效果。

let addFromEventObj = fromEvent(addBtn, 'click')
addFromEventObj = addFromEventObj.pipe(throttleTime(1000 * 10))
addFromEventObj = addFromEventObj.subscribe(() => {
nameInput.value = +(nameInput.value) + 1
})
複製程式碼

總結

上面我們已經簡單的分析了Rxjs 的基本使用方式，下面是我們這篇文章涉及的幾個基本概念

(Observable)可觀察的物件, (subscriber)觀察者, (pipe)管道, (throttleTime)操作符 下面我們根據這幾個概念來總結下Rxjs基於流來處理資料的整個過程(被觀察物件-> 資料處理-> 觀察者)

1. 首先需要建立一個可觀察物件， 在 esm2015\internal\observable\ 資料夾下都是用來建立可觀察物件的方法， 如上面我們Demo 用到的 fromEvent 就是其中一個，然後返回一個 Observable 物件， Observable 還有一個靜態方法 create 可以直接建立一個個 Observable 物件
2. pipe 就是將操作符掛載在 Observable 物件的 operator 屬性上， 如果pipe 傳遞了多個操作符， 則在 source 屬性(也是一個 Observable )物件的 operator ， 層層遞迴，從右到左

addFromEventObj = addFromEventObj.pipe(throttleTime(1000 * 2),mapTo(1), scan((init, next) => init + next, 0))
複製程式碼

其對應的物件如下圖：

3. 然後呼叫subscribe方法， 裡面的第一個引數就是觀察者,如下的subscriber 方法。

let addFromEventObj = fromEvent(addBtn, 'click')
addFromEventObj = addFromEventObj.pipe(throttleTime(1000 * 2),mapTo(1), scan((init, next) => init + next, 0))
const subscriber = value => {
nameInput.value = value
}
addFromEventObj = addFromEventObj.subscribe(subscriber)
複製程式碼