RxJS的另外四种实现方式（三）——性能最高的库

接上篇 RxJS的另外四种实现方式（二）——代码最小的库（续）

代码最小的库rx4rx-lite虽然在性能测试中超过了callbag，但和most库较量的时候却落败了，于是我下载了most库，要解开most库性能高的原因。 我们先上一组测试数据，这是在我的windows10 上面跑的

dataflow for 1000000 source events

lib

op/s

samples

rx4rx-lite

11.29 op/s ± 1.47%

(56 samples)

rx4rx-fast

22.56 op/s ± 1.77%

(57 samples)

cb-basics

9.56 op/s ± 1.73%

(49 samples)

xstream

5.37 op/s ± 0.68%

(30 samples)

most

17.32 op/s ± 1.93%

(82 samples)

rx 6

6.28 op/s ± 3.10%

(35 samples)

经过我的不懈努力终于把性能超过了most库。 我先介绍一下fast库的工作原理，下一篇文章我再介绍如何从most库中找到性能提升的要领。

在fast库中，我们开始使用一个基类作为一切操作符的父类，名为Sink。

1class Sink {
2    constructor(sink, ...args) {
3        this.defers = new Set()//用于存放需要释放的操作
4        this.sink = sink
5        this.init(...args)
6        if (sink) sink.defers.add(this)//用于释放的连锁反应
7    }
8    init() {
9
10    }
11	//是否连锁释放
12    set disposePass(value) {
13        if (!this.sink) return
14        if (value)
15            this.sink.defers.add(this)
16        else this.sink.defers.delete(this)
17    }
18	//数据向下传递
19    next(data) {
20        this.sink && this.sink.next(data)
21    }
22	//完成/error事件向下传递
23    complete(err) {
24        this.sink && this.sink.complete(err)
25        this.dispose(false)
26    }
27    error(err) {
28        this.complete(err)
29    }
30	//释放即取消订阅功能
31    dispose(defer = true) {
32        this.disposed = true
33        this.complete = noop
34        this.next = noop
35        this.dispose = noop
36        this.subscribes = this.subscribe = noop
37        defer && this.defer() //销毁时终止事件源
38    }
39    defer(add) {
40        if (add) {
41            this.defers.add(add)
42        } else {
43            this.defers.forEach(defer => {
44                switch (true) {
45                    case defer.dispose != void 0:
46                        defer.dispose()
47                        break;
48                    case typeof defer == 'function':
49                        defer()
50                        break
51                    case defer.length > 0:
52                        let [f, thisArg, ...args] = defer
53                        if (f.call)
54                            f.call(thisArg, ...args)
55                        else f(...args)
56                        break
57                }
58            })
59            this.defers.clear()
60        }
61    }
62    subscribe(source) {
63        source(this)
64        return this
65    }
66    subscribes(sources) {
67        sources.forEach(source => source(this))
68    }
69}

为了性能，代码量稍微有点多了。原本传入next和complete函数，现在变为传入sink对象，这里十分类似向Observable传入Observer对象。但是与rxjs不同的是，我们的Observable仍然是一个函数，我们看一个从数组构造Observable的代码

1exports.fromArray = array => sink => {
2    sink.pos = 0
3    const l = array.length
4    while (sink.pos < l && !sink.disposed) 
5		sink.next(array[sink.pos++])
6    sink.complete()
7}

这个pos为什么不直接定义一个变量呢？let pos = 0这是常规做法，这里把变量定义到了对象的属性上面，纯粹是为了提高一点点性能，经过测试发现，直接访问（读写操作）局部变量，比访问对象的属性要慢一些。

由于大部分的操作符都是相同的调用方式，所以可以抽象成一个函数

exports.deliver = Class => (...args) => source => sink => source(new Class(sink, ...args))

take操作符就变成了这样

1class Take extends Sink {
2    init(count) {
3        this.count = count
4    }
5    next(data) {
6        this.sink.next(data)
7        if (--this.count === 0) {
8            this.defer()
9            this.complete()
10        }
11    }
12}
13exports.take = deliver(Take)

而我们的subscriber就变成了这样

1exports.subscribe = (n, e = noop, c = noop) => source => {
2        const sink = new Sink()
3        sink.next = n
4        sink.complete = err => err ? e(err) : c()
5        source(sink)
6        return sink
7    }

至此fast库的基本构建逻辑已经展示完毕。 至于为什么这么快，就请听下回分解。 （未完待续）