[譯] 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 內部以外的地方也可使用。

若是你也喜歡本文，能夠考慮將此推文分享給你的關注者。