rxjs 源码分析1-(fromEvent)

时间  2019-11-21
标签 rxjs 源码 分析 fromevent 繁體版
原文   原文链接

前言

Rxjs是使用 Observables 的响应式编程的库，它使编写异步或基于回调的代码更容易。咱们如今针对Rxjs 6 来进行源码分析，分析其实现的基本原理， 咱们能够根据中文文档来学习Rxjs 的基本使用，可是这个文档是Rxjs 5 的版本。其最基本的使用区别以下,Rxjs 6的操做符都放在pipe (管道)中配置，而Rxjs 5 的版本是直接调用javascript

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 方法， 其最简单的一个范例以下：java

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
    })
复制代码

其实现的效果以下图：git

咱们点击加号或者减号 button 去给Input赋值。

原理

从上面咱们已经能够知道怎么去简单使用fromEvent, 下面咱们根据源代码来一步步深刻分析，其基本原理：github

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
})
复制代码

咱们将以下代码分红了两个步骤，bash

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 对象.async

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
})
复制代码
  1. 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))
复制代码

其对应的对象以下图：

  1. 而后调用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)
复制代码
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程