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 的父类。
0 Comments