Schedulers в RxJS

Что вы знаете о Schedulers в RxJS? Они скрывают от разработчиков работу с контекстом выполнения Observable. Как те эльфы-домовики из Гарри Поттера, которые выполняют всю черную работу в Хогвартсе, а о них никто даже и не слышал. Давайте исправим это и узнаем о них чуть больше.

Что такое Scheduler

Scheduler позволяет определить в каком контексте выполнения Observable будет доставлять нотификации до Observer. (вольный перевод документации)

Другими словами, Scheduler управляет очередностью и временем выполнения операций в Observable. Пример ниже поможет разобраться нам, как это работает.

import { of } from "rxjs";  console.log("Start"); of("Observable").subscribe(console.log); console.log("End");  // Logs: // Start // Observable // End

Два console.log зажали Observable между собой. Код выполняется синхронно. Если мы хотим, чтобы наш Observable выполнялся асинхронно, нужно добавить оператор observeOn и внутрь него прокинуть нужный нам Scheduler.

import { asyncScheduler, of } from "rxjs"; import { observeOn } from "rxjs/operators";  console.log("Start"); of("Observable")   .pipe(observeOn(asyncScheduler))   .subscribe(console.log); console.log("End");  // Logs: // Start // End // Observable

StackBlitz

Теперь наш Observable отдает данные асинхронно. Отлично. Это похоже, если бы мы обернули его в setTimeout(…, 0), не правда ли?

Как вы уже успели заметить, мы использовали asyncScheduler в нашем коде. Это один из легендарных Schedulers. Но их гораздо больше.

Типы Schedulers

Для понимания всей мощи Schedulers нужно знать, как работает Event Loop в JavaScript. Освежить знания можно с помощью этого видео или этой статьи на хабре.

Если кратко, то в браузере свой порядок выполнения кода:

1. Сначала выполняется синхронный код (callstack)

2. Дальше очередь микрозадач (Promise)

3. Потом очередь макрозадач (setTimeout, setInterval, XMLHttpRequest и т.д.).

4. Отдельно стоит очередь для задач, которые выполняются сразу перед следующим циклом перерисовки контента. (requestAnimationFrame)

В RxJS есть Scheduler на каждый из этих пунктов:

Еще существуют VirtualTimeScheduler и TestScheduler, которые используются для тестов. О них читайте здесь.

Взгляните на код ниже.

import { of, merge, asapScheduler,asyncScheduler,queueScheduler, animationFrameScheduler } from "rxjs"; import { observeOn } from "rxjs/operators";  const async$ = of("asyncScheduler").pipe(observeOn(asyncScheduler)); const asap$ = of("asapScheduler").pipe(observeOn(asapScheduler)); const queue$ = of("queueScheduler").pipe(observeOn(queueScheduler)); const animationFrame$ = of("animationFrameScheduler").pipe(   observeOn(animationFrameScheduler) ); merge(async$, asap$, queue$, animationFrame$).subscribe(console.log);  console.log("synchronous code");  // Logs: // queueScheduler // synchronous code // asapScheduler // animationFrameScheduler // asyncScheduler

StackBlitz

Как вы видите, «queueScheduler» отработал синхронно, так как он перед «synchronous code». А «asapScheduler» раньше «asyncScheduler», потому что в нем используется очередь микрозадач.

Как использовать Schedulers

Scheduler используется с операторами observeOn и subscribeOn. Оба принимают в себя первым аргументом Scheduler, а вторым аргументом delay, который по умолчанию равен нулю.

import { of, asyncScheduler } from "rxjs"; import { observeOn, subscribeOn } from "rxjs/operators";  of("observeOn")   .pipe(observeOn(asyncScheduler, 100))   .subscribe(console.log);  of("subscribeOn")   .pipe(subscribeOn(asyncScheduler, 50))   .subscribe(console.log);  // Logs: // subscribeOn // observeOn

StackBlitz

Различие их в том, что observeOn планирует в каком контексте будут выполняться методы observer — next, error и complete выполняются в соответствующем с Scheduler контексте. А subscribeOn влияет на subscriber — метод subscribe будет выполняться в другом контексте.

Интересный факт, что если задать delay не равный нулю в observeOn/subscribeOn, то вне зависимости какой используется Scheduler, будет использоваться asyncScheduler. Бессмысленный код — observeOn(animationFrameScheduler, 100).

До версии RxJS 6.5.0 можно было добавить Scheduler вторым аргументом для of, from, merge, range и т.д. В новых версиях RxJS это поведение deprecated, и необходимо использовать функцию scheduled для этого.

import { of, scheduled, asapScheduler } from 'rxjs';  // DEPRECATED // of(2, asapScheduler).subscribe(console.log);  scheduled(of('scheduled'), asapScheduler).subscribe(console.log);

Пример использования Scheduler

Мы не задумываемся о Schedulers при работе с RxJS, так как авторы библиотеки провели великолепную работу по абстрагированию этой логики. Но бывает, когда использование Scheduler будет органично, и вы должны быть к этому готовы. На моей практике был один такой случай. Меня попросили реализовать кэширование запросов отличающихся по индетификатору. Примерно такую функцию я написал:

const cache = new Map<number, any>(); function get(id: number): Observable<any> {  if (cache.has(id)) {    return of(cache.get(id));  }  return http.get(‘some-url\’ + id).pipe(    tap(data => {      cache.set(id, data);    }),  ); } 

Код работает. Но есть один нюанс. В первый раз, когда использовали мою функцию, она возвращала значение асинхронно, а во второй раз — синхронно. Когда функция ведет себя так непресказуемо для вызывающего, может случится трагедия. Выходит, что я высвободил Залго.

Добавим scheduled с asyncScheduler в 4 строчке, чтобы исправить это.

return scheduled(of(cache.get(id)), asyncScheduler);

Теперь все работает предсказуемо. Залго низвержен туда, где был.

Заключение

Schedulers влияют на время и порядок выполнения задач. Четверть операторов RxJS использует их под капотом. С большой вероятностью на практике осведомленность о них не нужна. Но никогда не знаешь, в какой момент жизни пригодится Scheduler.

Я буду рад услышать о вашем опыте использования Schedulers в комментариях. Спасибо за внимание!