[译] RxJS 调度器入门

原文连接: staltz.com/primer-on-r…
本文为 RxJS 中文社区 翻译文章，如需转载，请注明出处，谢谢合做！
若是你也想和咱们一块儿，翻译更多优质的 RxJS 文章以奉献给你们，请点击【这里】

RxJS 中的调度器 ( Schedulers ) 是用来控制事件发出的顺序和速度的(发送给观察者的)。它还能够控制订阅 ( Subscriptions ) 的顺序。为了避免搞得太理论化，先考虑下这个示例:

const a$ = Rx.Observable.of(1, 2);
const b$ = Rx.Observable.of(10);

const c$ = Rx.Observable.combineLatest(a$, b$, (a, b) => a + b);

c$.subscribe(c => console.log(c));复制代码

你以为控制台输出的结果是什么呢？大多数人会猜是:

11
12复制代码

由于首先 a$ 中1会和 b$ 中的10配对，而后 a$ 中的2和 b$ 中的10配对。

事实上，出如今控制台中的是:

12复制代码

1 + 10 的组合并无发生。缘由是 Observables a$ 和 b$ 都是“同步的”，它们会尽量快地执行。那么事件发出的顺序究竟是怎样的呢？答案是不肯定的，它多是如下任意一种:

• 1, 2, 10
• 1, 10, 2
• 10, 1, 2

在这种顺序不肯定的状况下，咱们应该描述出事件的发出顺序是怎样的。这就是调度器所作的事。默认状况下，RxJS 使用所谓的递归调度器。下面是它的工做原理:

1. c$ 被订阅
2. combineLatest 的第一个输入流 a$ 被订阅
3. a$ 发出值 1
4. combineLatest 将 1 做为 a$ 的最新值进行保存
5. a$ 发出值 2
6. combineLatest 将 2 做为 a$ 的最新值进行保存
7. combineLatest 的第二个输入流 b$ 被订阅
8. b$ 发出值 10
9. combineLatest 将 10 做为 b$ 的最新值进行保存
10. combineLatest 如今同时拥有了 a$ 和 b$ 的值，所以它发出值 2 + 10

发出的顺序为 1, 2, 10 。最有意思的是在 b$ 被订阅前， 将 a$ 的全部事件都尽快地发出了。RxJS 使用这种调度器做为默认调度器出于两点缘由:

• 使用此策略性能的总体表现更好
• 在调试工具中更易于进行堆栈跟踪

然而，能够经过使用不一样的调度器来自定义事件发出的顺序及速度。咱们在 a$ 上使用 asap 调度器来让其“慢下来”:

// const a$ = Rx.Observable.of(1, 2);
const a$ = Rx.Observable.from([1, 2], Rx.Scheduler.asap); // 新代码
const b$ = Rx.Observable.of(10);

const c$ = Rx.Observable.combineLatest(a$, b$, (a, b) => a + b);

c$.subscribe(c => console.log(c))复制代码

from 操做符的第二个参数是调度器，用来自定义事件的发出。asap 调度器使用 setImmediate 来安排任务尽快运行，但不是同步的。代码改变后，控制台会输出:

11
12复制代码

由于内部运行顺序以下:

1. c$ 被订阅
2. combineLatest 的第一个输入流 a$ 被订阅
3. combineLatest 的第二个输入流 b$ 被订阅
4. b$ 发出值 10
5. combineLatest 将 10 做为 b$ 的最新值进行保存
6. a$ 发出值 1
7. combineLatest 将 1 做为 a$ 的最新值进行保存
8. combineLatest 如今同时拥有了 a$ 和 b$ 的值，所以它发出值 1 + 10
9. a$ 发出值 2
10. combineLatest 将 2 做为 a$ 的最新值进行保存
11. combineLatest 发出值 2 + 10

发出的顺序为 10, 1, 2 。为了获得另一种发出顺序，能够为 b$ 也自定义调度器:

const a$ = Rx.Observable.from([1, 2], Rx.Scheduler.asap);
// const b$ = Rx.Observable.of(10);
const b$ = Rx.Observable.from([10], Rx.Scheduler.asap); // 新代码

const c$ = Rx.Observable.combineLatest(a$, b$, (a, b) => a + b);

c$.subscribe(c => console.log(c));复制代码

如今发出的顺序为 1, 10, 2，由于运行顺序以下:

1. c$ 被订阅
2. combineLatest 的第一个输入流 a$ 被订阅
3. combineLatest 的第二个输入流 b$ 被订阅
4. a$ 发出值 1
5. combineLatest 将 1 做为 a$ 的最新值进行保存
6. b$ 发出值 10
7. combineLatest 将 10 做为 b$ 的最新值进行保存
8. combineLatest 如今同时拥有了 a$ 和 b$ 的值，所以它发出值 1 + 10
9. a$ 发出值 2
10. combineLatest 将 2 做为 a$ 的最新值进行保存
11. combineLatest 发出值 2 + 10

调度器还可让事件的发出变得更快，同时保持发出的顺序不变。例如，RxJS 的 TestScheduler 可使 Observable.interval(1000).take(10) 被订阅时进行同步执行，而不须要花费10秒钟来完成:

Rx.Observable.interval(1000, new Rx.TestScheduler()).take(10)复制代码

TestScheduler 是在 RxJS 内部使用的 (参见 filter 的测试用例)，它使得成百上千个时间相关的测试代码飞快地运行，但有一些像 Rx Sandbox 这样的工具和积极的讨论来丰富此调度器的使用场景，使得在 RxJS 内部以外的地方也可使用。

若是你也喜欢本文，能够考虑将此推文分享给你的关注者。