80 行代碼實現簡易 RxJS

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：

觀察下它的特點：

它接收一個回調函數，裏面可以調用 next 來傳輸數據。
它有 subscribe 方法可以用來添加 Observer 的訂閱，返回 subscription
它可以在回調函數里返回 unsbscribe 時的處理邏輯
它有 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 行代碼。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/G-Af4gDuBuXuHMTV-de-0w

RxJS 的使用

80 行代碼實現 RxJS

總結

猜你喜歡