Эффективность RxJS: управляем производительностью и оптимизируем подписки / Хабр

Введение

RxJS — один из самых мощных инструментов в экосистеме Angular, который позволяет работать с асинхронными потоками данных с элегантностью и гибкостью. Однако за этой мощью скрывается коварная проблема: при неправильном управлении потоками ваши приложения смогут утонуть в утечках памяти. Это — потенциально незаметный убийца производительности. В этой статье я, постараюсь поделиться практиками управления потоками RxJS и посоветую, как избежать утечек памяти.

Чем опасны утечки в потоках RxJS?

RxJS позволяет легко подписаться на потоки данных с помощью оператора subscribe. Однако каждая подписка, подобно таймеру или обработчику событий, живёт до тех пор, пока её явно не отменить или поток не завершится. Если в компоненте Angular подписка остаётся активной после того, как компонент уничтожен, то:

Данные будут продолжать поступать в уничтоженный компонент.
Зависимые ресурсы (например, ссылки на DOM-узлы или другие потоки) не смогут быть освобождены.
Это приведёт к утечкам памяти, что особенно критично в крупных приложениях с множеством пользователей.

Холодные и горячие потоки

Холодные потоки

Холодные потоки создают уникальную цепочку для каждого подписчика. Это значит, что каждый раз, когда вы подписываетесь на поток, он заново запускается, что может приводить к выполнению одних и тех же операций, например, API-запросов или тяжёлых вычислений.

Пример:

import { Observable } from 'rxjs';

const cold$ = new Observable(observer => {
  console.log('Поток запущен');
  observer.next(Math.random());
  observer.complete();
});

cold$.subscribe(value => console.log('Подписчик 1:', value));
cold$.subscribe(value => console.log('Подписчик 2:', value));
// Вывод в консоль:
// Поток запущен
// Подписчик 1: 0.123...
// Поток запущен
// Подписчик 2: 0.456...

Каждый подписчик получает собственную уникальную копию потока.

Горячие потоки

Горячий поток уже запущен и работает независимо от подписчиков. Все подписчики получают данные одновременно, но если кто-то подписывается позже, он пропустит начальные события.

Пример:

import { Subject } from 'rxjs';

const hot$ = new Subject<number>();

hot$.subscribe(value => console.log('Подписчик 1 получает:', value));

hot$.next(42); // Данные отправляются текущему подписчику

hot$.subscribe(value => console.log('Подписчик 2 получает:', value));
// Вывод:
// Подписчик 1 получает: 42

Здесь подписчик 2 пропускает первое значение (42), так как оно было отправлено до того, как он подписался.

Как превратить холодный поток в горячий?

Используйте операторы share или shareReplay, чтобы сделать поток горячим.

import { interval } from 'rxjs';
import { share } from 'rxjs/operators';

const cold$ = interval(1000).pipe(share());

cold$.subscribe(value => console.log('Подписчик 1:', value));
setTimeout(() => cold$.subscribe(value => console.log('Подписчик 2:', value)), 3000);
// Вывод:
// Подписчик 1: 0
// Подписчик 1: 1
// Подписчик 1: 2
// Подписчик 1: 3
// Подписчик 2: 3

Теперь поток сохраняет состояние и становится доступным для всех подписчиков.

Стратегии управления потоками RxJS

Теперь давайте посмотрим, как мастерски справляться с потоками и исключать любые утечки.

1. Правильное использование takeUntil

Вместо того чтобы вручную отслеживать и отменять подписки, используйте поток-сигнал завершения. Обычно это реализация через Subject, который эмитит значение перед уничтожением компонента.

Пример:

import { Component, OnDestroy } from '@angular/core';
import { Subject, interval, takeUntil } from 'rxjs';

@Component({
  selector: 'app-example',
  template: `<p>Поток работает! Откройте консоль.</p>`
})
export class ExampleComponent implements OnDestroy {
  private destroy$ = new Subject<void>();

  ngOnInit() {
    interval(1000) // каждую секунду
      .pipe(takeUntil(this.destroy$)) // завершение потока при уничтожении компонента
      .subscribe(value => console.log(value));
  }

  ngOnDestroy() {
    this.destroy$.next(); // отправляем сигнал завершения
    this.destroy$.complete(); // очищаем destroy$
  }
}

Используйте takeUntil в комбинации с Angular Lifecycle hooks (ngOnDestroy), чтобы гарантировать очистку потоков.

2. Используйте AsyncPipe — минимизируйте подписки вручную

AsyncPipe автоматически управляет подписками и отписками в шаблонах. Используйте его, когда данные передаются из компонентов или сервисов в UI, чтобы не писать лишний код.

Пример (без дополнительного кода отписки):

<div *ngIf="data$ | async as data">
  {{ data }}
</div>

Преимущества:

Автоматическая отписка, когда компонент уничтожается.
Меньше кода — меньше вероятность ошибок.

3. Ограничьте подписки с помощью операторов take, first или takeWhile

Если вам нужны только первые несколько значений из потока, ограничьте подписку операторами. Например, take(1) автоматически завершит поток после первого значения.

Пример:

this.someService.getData().pipe(
  take(1)
).subscribe(data => console.log(data));

Таким образом, вы избегаете вечного слушателя и облегчаете управление потоками.

4. Использование Subscription и add()

Если подписки создаются вручную через subscribe, объединяйте их в один Subscription объект и очищайте их все одновременно.

Пример:

import { Subscription } from 'rxjs';

@Component({
  selector: 'app-example',
  template: `<p>Пример с Subscription</p>`
})
export class ExampleComponent implements OnDestroy {
  private subscriptions = new Subscription();

  ngOnInit() {
    const sub1 = this.service.getData1().subscribe();
    const sub2 = this.service.getData2().subscribe();

    this.subscriptions.add(sub1);
    this.subscriptions.add(sub2);
  }

  ngOnDestroy() {
    this.subscriptions.unsubscribe();
  }
}

Для упрощения работы можно создать абстрактный класс, содержащий механизмы onDestroy & поле subscriptions, и наследовать все компоненты от него, чтобы избежать муторного дублирования кода.

5. Ленивая загрузка потоков (switchMap)

Когда компоненту нужно реагировать на изменения одного потока, трансформируя его в другой, используйте switchMap вместо ручной подписки на вложенные потоки. Так вы гарантируете, что только последний поток активен, а предыдущие автоматически завершаются.

Пример (избежание гонок данных):

this.searchControl.valueChanges.pipe(
  switchMap(searchTerm => this.api.searchByTerm(searchTerm))
).subscribe(results => this.searchResults = results);

6. Единая подписка с shareReplay.

Если несколько подписчиков используют один и тот же поток, важно не создавать дополнительных операций. Благодаря shareReplay данные кешируются и передаются повторно.

import { timer } from 'rxjs';
import { shareReplay } from 'rxjs/operators';

const apiCall$ = timer(2000).pipe(shareReplay(1));

apiCall$.subscribe(value => console.log('Подписчик 1:', value));
apiCall$.subscribe(value => console.log('Подписчик 2:', value));

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

7. Ленивая инициализация потоков с defer.

Иногда поток нужно запускать не сразу, а только в момент подписки. Это полезно для тяжёлых операций.

import { defer, timer } from 'rxjs';

const lazy$ = defer(() => {
  console.log('Инициализация потоков...');
  return timer(1000);
});

lazy$.subscribe(value => console.log(value));

8. "Прореживание" событий с throttleTime.

Часто пользовательские события создаются слишком часто (например, события scroll, mousemove). Для их оптимизации можно использовать throttleTime:

import { fromEvent } from 'rxjs';
import { throttleTime } from 'rxjs/operators';

const scroll$ = fromEvent(document, 'scroll').pipe(throttleTime(200));

scroll$.subscribe(() => console.log('Событие Scroll обработано!'));

9. Завершение потоков с помощью takeUntilDestroyed

*Добавлено по предложению Ant0ha

takeUntilDestroyed — это относительно современный подход (Angular 16+), предназначенный для автоматического завершения потоков при уничтожении компонента, директивы или сервиса.

import {Component, DestroyRef, inject, Injector, OnInit} from '@angular/core';
import {takeUntilDestroyed} from "@angular/core/rxjs-interop";
import {interval} from "rxjs";

@Component({
  selector: 'app-destr',
  standalone: true,
  templateUrl: './destr.component.html',
  styleUrl: './destr.component.scss'
})
export class DestrComponent implements OnInit{
  private readonly destroyRef = inject(DestroyRef);

  ngOnInit() {
    interval(1000).pipe(
      takeUntilDestroyed(this.destroyRef) // Автоматическая отписка
    ).subscribe(value => console.log(value));
  }
}

*Для любителей работать с подписками из конструктора: там можно просто вызвать takeUntilDestroyed(), контекст компонента подхватится автоматически.

AsyncPipe в Angular

В Angular вы можете автоматически обрабатывать подписки, используя AsyncPipe. Это не только упрощает код, но и предотвращает утечки памяти.

Шаблон:

@Component({
  template: `<div *ngIf="data$ | async as data">{{ data }}</div>`,
})
export class ExampleComponent {
  data$ = this.http.get('/api/data');
}

Инструменты для мониторинга утечек

Chrome DevTools: используйте вкладку Performance и таймлайны для выявления аномалий в памяти.
RxJS Marbles: полезно для тестирования потоков.

Советы по оптимизации потоков

Не забывайте завершать потоки. Используйте unsubscribe, takeUntil или AsyncPipe.
Избегайте лишних подписок. Старайтесь объединять подписчиков и делиться данными через share и shareReplay.
Профилируйте потоки с помощью tap. Вставляйте tap для проверки данных во время разработки.
Минимизируйте сложные операции. Используйте прореживание событий (throttleTime), ограничение (take) и фильтрацию (filter).

Заключение

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