一文带你用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：

观察下它的特点：

  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(() =>             

                本文名称：一文带你用80行代码实现简易RxJS
                

                分享URL：http://www.mswzjz.cn/qtweb/news40/123190.html
            

            

                攀枝花网站建设、攀枝花网站运维推广公司-贝锐智能，是专注品牌与效果的网络营销公司；服务项目有等
                                

            

                广告
                
            

            

                声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：
                贝锐智能

贝锐智能技术为您推荐以下文章

网站收录知识

分类信息网

成都平面广告设计　　 　佑馨产后护理　　 　草坪音箱　　 　云服务器　　 　温江电信机房　　 　成都小程序开发　　 　成都网站优化　　 　定制网站设计　　 　乐山服务器托管　　 　金羊影视传媒　　 　成都户外广告　　 　服务器租凭　　 　梦港之夜　　 　成都app软件开发　　 　芦山网站建设　　 　成都服务器托管　　 　营山月子护理　　 　成都服务器机柜租用　　 　微信公众号开发　　 　成都网站建设