Четвертый шаг в мир RxJs: незавершенные потоки — тихие убийцы приложений

• Первый шаг в мир RxJS: знакомство с Observables

• Второй шаг в мир RxJS: Операторы RxJS — как изучать и зачем они нужны

• Третий шаг в мир RxJS: комбинирование потоков в RxJS

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

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

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

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

Угрозы при работе с RxJS

Без RxJS сегодня сложно представить современный Angular. Она помогает нам справляться с асинхронными задачами, реактивно управлять состоянием, реагировать на события и даже строить сложные цепочки обработки данных.

С RxJS вы работаете с потоками данных (Observable). Когда вы подписываетесь на Observable, можно представить это как подключение шланга к крану с водой. Поток данных начинает течь, события льются в приложение непрерывным потоком. Проблема в том, что кран не закроется, пока вы сами не повернёте ручку.

Подписки без unsubscribe — тихие убийцы. Пока вы тестируете всё в процессе разработки небольшими сессиями, приложение работает идеально… а потом тестировщик жалуется на дым из ноутбука после 30 минут общения с результатом вашей работы.

Представьте, что вы сделали подписку внутри компонента Angular, но забыли её «закрыть», когда компонент больше не нужен (например, пользователь ушел в другой раздел приложения). Что будет происходить?

• Данные всё еще текут. Даже если UI больше не использует эти данные, подписка остается активной. Это значит, даже если никто не смотрит, поток продолжает передавать события, загружая и вашу систему, и сервер.

• Память перестает очищаться. RxJS потоки работают как долгоживущие объекты. Пока подписка активна, ссылка на данные сохраняется в памяти, а сборщик мусора (GC) не сможет их освободить.

• Незаметные проблемы становятся серьёзными. На первых порах это может казаться неважным — всё работает. Но через пару часов работы приложения (или после нескольких переходов между компонентами) проблемы начинают накапливаться:

  • Падение производительности.

  • Утечка памяти.

  • Краш браузера (особенно на слабых устройствах).

Почему вы не заметите проблему сразу? В процессе разработки легко упустить из виду детали подписок в RxJS. Все выглядит хорошо: данные обновляются, ошибки в консоли отсутствуют, скорость в коротких сессиях нормальная. Но проблема с неоптимальной работой всегда догоняет. Первая серьезная утечка памяти может поймать вас спустя дни и недели, когда приложение попадёт в руки пользователю со слабым устройством. И тогда вам придется пройти все круги ада с поиском «плавающих» багов, ошибок окружения, сетования на кривые пользовательские руки…

Пример с утечками памяти

Вообразим, что необходимо разработать некий чат для сайта. Он получает сообщения в режиме реального времени и отображает их на странице. Всё выглядит прекрасно, но внутри есть ошибка. Настолько крохотная, что её сложно заметить… до первого отчёта о деградации производительности, которая приходит к вам прямиком от пользователя.

import {ChangeDetectionStrategy, ChangeDetectorRef, Component, OnDestroy, OnInit} from '@angular/core'; import {interval, map} from "rxjs"; import {NgForOf} from "@angular/common";  let instance = 1;  @Component({    selector: 'app-chat',    standalone: true,    imports: [       NgForOf    ],    changeDetection: ChangeDetectionStrategy.OnPush,    template: `     <h2>Real-time Chat</h2>     <div class="chat-window">       <div *ngFor="let message of messages" class="message">         {{ message }}       </div>     </div>   `,    styles: [`     .chat-window {       border: 1px solid #ccc;       height: 300px;       overflow-y: auto;     }     .message {       padding: 8px;       border-bottom: 1px solid #eee;     }   `] }) export class ChatComponent implements OnInit, OnDestroy {    messages: string[] = [];    instanceId = instance++;    // создание поддельной строки    private getRandomString = () => (Math.random() + 1).toString(36).substring(2);     constructor(private detector: ChangeDetectorRef) {    }     ngOnInit(): void {       // Создаем поток новых сообщений каждую половину секунды       interval(500)               .pipe(                       map((i) => `New message #${i + 1} - ${new Array(1000).fill(0)                               .map(() => this.getRandomString()).join('')}`) // Симуляция тяжелых данных               )               .subscribe(message => {                  // Добавляем новое сообщение в список                  this.messages.push(message);                  this.detector.detectChanges();                  console.log(`Подписка ${this.instanceId} отработала.`);               });    }      ngOnDestroy(): void {       console.log(`ChatComponent ${this.instanceId} destroyed`);    } }

Что не так с этим компонентом?

На первый взгляд, ничего катастрофического, UI обновляется, всё работает как задумывалось.

Но вот проблема: вы НЕ отписались от потока!

Когда пользователь скрывает чат, компонент ChatComponent уничтожается. Но подписка, которая срабатывает два раза в секунду, продолжает «жить». Она продолжает генерировать сообщения, чтобы каждый из них отправить… никуда. В результате:

• Поток всё ещё активен. Сообщения продолжают генерироваться в памяти.

• Ссылки на массив messages[] остаются в памяти.

• Сборщик мусора (GC) не сможет освободить занятые ресурсы, потому что подписка поддерживает «живую» ссылку.

Давайте посмотрим на это в жизни:

• Скачайте проект отсюда и запустите.

• Включите DevTools браузера и перейдите в раздел Console.

• Понажимайте на кнопку открытия/закрытия чата несколько раз.

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

Почему это плохо? Утечки памяти и их влияние

Когда вы подписываетесь на Observable-поток, он создаёт ссылку на ваш код-подписчик. RxJS поток продолжает «жить» до тех пор, пока есть хотя бы одна активная подписка. Это значит, что если вы не отписались, задействованные данные будут оставаться в памяти бесконечно долго.

В нашем примере подписка сохраняет ссылку на массив messages[]. Даже если пользователь закрывает чат, память не освобождается:

• Генерируется очередной message, добавляется в массив, но он уже никому не нужен.

• Старая ссылка на массив остаётся «живой», а сборщик мусора (GC) не может её освободить, потому что поток всё ещё активен.

Результат? Память начинает «забиваться». На ранних этапах этого незаметно, но через 5-10 минут работы приложение вдруг начнёт тормозить.

Чем больше подписок остаётся активными, тем больше данных обрабатывает ваш поток RxJS. В результате:

• ЦП начинает загружаться.

• Слабые устройства (например, старые телефоны) с трудом справляются с лишними вычислениями.

• В самый неподходящий момент браузер «ложится» или появляется серьёзная задержка графического интерфейса.

В случае с нашим чатом баг особенно опасен, потому что новые сообщения генерируются каждые 500 мс. Умножьте это на количество оставленных подписок — и вы получите катастрофу.

Почему это неочевидно?

Ошибки, связанные с утечками памяти, часто развиваются постепенно. Вот несколько причин, почему многие забывают про unsubscribe на ранних этапах разработки:

• Производительность на старте нормальная. В начале вы ничего не замечаете — подписка работает, UI обновляется.

• Сложность диагностики. Утечки памяти проявляются только в долгосрочной перспективе. Вам редко удаётся «увидеть» что-то странное, пока приложение работает всего несколько минут.

• Angular сам справляется с многими задачами за вас. Инструменты фреймворка часто «маскируют» проблему до тех пор, пока она не начнёт накапливаться.

• Оптимистичное мышление. «Работает — значит, всё нормально». Но правда в том, что незакрытые подписки всегда найдут способ как создать вам проблему в самый неподходящий момент.

Как это исправить? Закрываем подписки

Итак, мы поняли, что оставлять подписки открытыми — плохая идея. Теперь давайте рассмотрим, как исправить эту проблему, используя несколько простых и надёжных методов. На самом деле, грамотно управлять подписками в вашем Angular-приложении не так сложно, как может показаться. Главное — сделать это правилом с самого начала.

1. Используйте единый подход к управлению подписками

1. Используем Subscription и ручное управление

RxJS предоставляет специальный класс Subscription, который помогает управлять подписками. Используя его, вы можете явно сохранять подписки, а затем отписываться, когда компонент больше не нужен. Давайте модифицируем наш проблемный код с использованием Subscription:

import {ChangeDetectionStrategy, ChangeDetectorRef, Component, OnDestroy, OnInit} from '@angular/core'; import {interval, map, Subscription} from "rxjs"; import {NgForOf} from "@angular/common";  let instance = 1;  @Component({     selector: 'app-chat',     standalone: true,     imports: [         NgForOf     ],     changeDetection: ChangeDetectionStrategy.OnPush,     template: `     <h2>Real-time Chat</h2>     <div class="chat-window">       <div *ngFor="let message of messages" class="message">         {{ message }}       </div>     </div>   `,     styles: [`     .chat-window {       border: 1px solid #ccc;       height: 300px;       overflow-y: auto;     }     .message {       padding: 8px;       border-bottom: 1px solid #eee;     }   `] }) export class ChatComponent implements OnInit, OnDestroy {     messages: string[] = [];     instanceId = instance++;     private subscriptions: Subscription[] = []; // Сохраняем все подписки здесь     private getRandomString = () => (Math.random() + 1).toString(36).substring(2);      constructor(private detector: ChangeDetectorRef) {     }      ngOnInit(): void {         // Добавляем наблюдаемый поток в список подписок         this.subscriptions.push(interval(500)             .pipe(                 map((i) => `New message #${i + 1} - ${new Array(1000)                     .fill(0).map(() => this.getRandomString()).join('')}`)             )             .subscribe(message => {                 // Добавляем новое сообщение в список                 this.messages.push(message);                 this.detector.detectChanges();                 console.log(`Подписка ${this.instanceId} отработала.`);             }));     }       ngOnDestroy(): void {         // Отписываемся от всех наблюдаемых потоков         this.subscriptions.forEach(s => s.unsubscribe());         console.log(`ChatComponent ${this.instanceId} destroyed`);     } }

Что здесь происходит?

• Мы создаём объект Subscription, куда добавляем наш поток сообщений.

• В момент уничтожения компонента (ngOnDestroy) вызываем .unsubscribe(), и завершаем наблюдение за потоком данных.

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

А еще можно создать специальный служебный класс, который содержит необходимые методы:

import { OnDestroy } from '@angular/core'; import { Subscription } from 'rxjs';  export abstract class DestructibleComponent implements OnDestroy {    protected subs: Subscription[] = [];    protected onDestroy?: () => void;     ngOnDestroy(): void {       this.subs.forEach(s => s.unsubscribe());       if (this.onDestroy) this.onDestroy();    } }

Любой компонент может унаследовать этому классу (extends) и добавлять свои подписки в массив subs, без необходимости определять ngOnDestroy. Если же ему понадобится совершить какие-то действия при уничтожении, то в наличии есть специальный метод onDestroy, созданный специально для этого и не влияющий на отписки.

2. Используем AsyncPipe: минимум кода

Если ваши данные отображаются только в шаблоне (например, через *ngFor или {{ }}), вы можете вообще обойтись без подписок в TypeScript. В этом случае используется Angular AsyncPipe, который сам заботится об отписке.

Вот пример:

import {ChangeDetectionStrategy, ChangeDetectorRef, Component, OnDestroy} from '@angular/core'; import {interval, map, tap} from "rxjs"; import {AsyncPipe, NgForOf} from "@angular/common";  let instance = 1;  @Component({    selector: 'app-chat',    standalone: true,    imports: [       NgForOf,       AsyncPipe    ],    changeDetection: ChangeDetectionStrategy.OnPush,    template: `     <h2>Real-time Chat</h2>     <div class="chat-window">       <div *ngFor="let message of messages$ | async" class="message">         {{ message }}       </div>     </div>   `,    styles: [`     .chat-window {       border: 1px solid #ccc;       height: 300px;       overflow-y: auto;     }     .message {       padding: 8px;       border-bottom: 1px solid #eee;     }   `] }) export class ChatComponent implements OnDestroy {    instanceId = instance++;    private getRandomString = () => (Math.random() + 1).toString(36).substring(2);    messages$ = interval(500)            .pipe(                    tap(() => console.log(`Подписка ${this.instanceId} отработала.`)),                    map((i) => Array.from({length: i}).map((_, idx) =>                            `New message #${idx + 1} - ${new Array(1000).fill(0)                                    .map(() => this.getRandomString()).join('')}`))            );     constructor(private detector: ChangeDetectorRef) {    }     ngOnDestroy(): void {       console.log(`ChatComponent ${this.instanceId} destroyed`);    } }

Почему AsyncPipe — это круто?

• Он автоматически подписывается на Observable, когда шаблон отображается.

• Отписка выполняется автоматически, когда компонент уничтожается.

• Код становится максимально лаконичным.

Когда это подходит?

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

Рекомендации для работы с подписками

1. Используйте единый подход к управлению подписками

Во всем проекте стоит придерживаться одного стиля работы с подписками. Например:

• Используйте наследование DestructibleComponent для всех компонентов.

• Используйте широко известный подход с takeUntil

• Или применяйте Subscription для ручного управления, если подписки более специфичны.

1. Используйте единый подход к управлению подписками

2. Минимизируйте количество подписок

Старайтесь комбинировать потоки с помощью операторов merge, combineLatest, switchMap, вместо создания нескольких небольших подписок.

3. Отдавайте предпочтение AsyncPipe

Если ваши данные отображаются только в шаблонах, используйте AsyncPipe.

4. Следите за производительностью в DevTools

Регулярно проводите анализ производительности приложения:

• Используйте вкладку Memory для проверки утечек памяти.

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

5. Логируйте уничтожение подписок в процессе разработки

Добавляйте console.log в ngOnDestroy, чтобы убедиться, что подписки закрываются при уничтожении компонента.

Заключение

Работа с RxJS — это не просто освоение нового инструмента. Это путь к более эффективному, организованному и глубокому пониманию разработки современных приложений. На этом пути неизбежно будут ошибки, как те самые незакрытые подписки, но каждая из них — это шаг вперёд, шаг к вашему профессиональному росту.

RxJS, как и любое другое сложное, но полезное умение, требует терпения, практики и упрямой решимости стать лучше.

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

Память браузера, пользователи и ваши коллеги скажут вам спасибо. Уверенно пробуйте, ошибайтесь, исправляйтесь, делайте выводы и двигайтесь к новым горизонтам. Успех в каждом аспекте разработки — это результат собственной работы и стремления превратить сегодняшние сложности в завтрашнее уверенное знание.