关于rxjs:80-行代码实现简易-RxJS

44次阅读

共计 7556 个字符，预计需要花费 19 分钟才能阅读完成。

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 行代码。

最初
如果你感觉此文对你有一丁点帮忙，点个赞。或者能够退出我的开发交换群：1025263163 互相学习，咱们会有业余的技术答疑解惑

如果你感觉这篇文章对你有点用的话，麻烦请给咱们的开源我的项目点点 star:http://github.crmeb.net/u/defu 不胜感激！

PHP 学习手册：https://doc.crmeb.com
技术交换论坛：https://q.crmeb.com

正文完
rxjs
发表至： rxjs
2022-02-28
 0
关于rxjs:使用-RxJs-实现一个支持-infinite-scroll-的-Angular-Component
关于rxjs:NgRx-里-first-和-take1-操作符的区别
关于rxjs:RxJs-fromEvent-工作原理分析
关于rxjs:RxJs-map-operator-工作原理分析