Що таке subject і які типи subject існують в RxJS?

Angular~4 хв читання

Subject в RxJS є одночасно Observable і Observer: він надсилає (мультикастить) значення всім активним підписникам через .next().

Теорія

TL;DR

Subject схожий на живу радіотрансляцію: підключився пізніше і пропустив те, що вже сказали
BehaviorSubject дає новому підписнику останнє значення одразу (підходить для стану)
ReplaySubject(n) буферизує останні n значень і відтворює їх для пізніх підписників
AsyncSubject видає тільки останнє значення після виклику .complete()
Вибір: не потрібна історія (Subject), поточний стан (BehaviorSubject), останні N подій (ReplaySubject), тільки фінальний результат (AsyncSubject)

Короткий приклад

typescript

import { Subject } from 'rxjs';

const messages$ = new Subject<string>();

// Обидва підписники з'єднуються до першого emit
messages$.subscribe(val => console.log('A:', val));
messages$.subscribe(val => console.log('B:', val));

messages$.next('Привіт');
// A: Привіт
// B: Привіт  <- одне значення, один виклик, два підписники

Один .next() доставив значення обом підписникам одночасно. Саме це і є мультикаст. Звичайний new Observable() створив би окреме виконання для кожного підписника.

Гарячий проти холодного: головна різниця

Звичайний Observable є холодним (cold): він запускає нове виконання для кожного підписника. Subject є гарячим (hot) за замовчуванням. Він працює незалежно від кількості підписників, а значення передаються вручну через .next(), а не всередині фабричної функції.

Всередині RxJS Subject розширює Observable і реалізує інтерфейс Observer. Він підтримує масив об'єктів Subscriber. Коли викликається .next(value), Subject синхронно проходить по масиву і викликає next(value) кожного підписника.

Типи Subject: швидкий огляд

Тип	Зберігає значення?	Новий підписник отримує	Типове застосування
`Subject`	Ні	Тільки майбутні виклики `.next()`	Живі події, кліки кнопок
`BehaviorSubject`	Останнє значення	Поточне значення одразу	Стан UI, статус авторизації
`ReplaySubject`	Останні N значень	Буферизовані значення, потім нові	Історія чату, стек скасувань
`AsyncSubject`	Останнє значення	Тільки після `complete()`	Одноразова відповідь API

BehaviorSubject

BehaviorSubject вимагає початкового значення і завжди зберігає найактуальніше. Будь-який підписник отримує це значення синхронно в момент підписки, ще до будь-яких нових emit.

typescript

import { Injectable } from '@angular/core';
import { BehaviorSubject } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class UserService {
  private statusSubject = new BehaviorSubject<'online' | 'offline'>('offline');
  public status$ = this.statusSubject.asObservable(); // приховує .next() від зовні

  setOnline() {
    this.statusSubject.next('online');
  }
}

// Компонент підписується і одразу отримує 'offline'
userService.status$.subscribe(s => console.log('Статус:', s));
// → Статус: offline  (синхронно, в момент підписки)

userService.setOnline();
// → Статус: online  (для всіх поточних підписників)

.asObservable() обгортає Subject у звичайний Observable і приховує .next() від споживачів. Тільки сервіс контролює зміни значення. Майже в кожному проєкті, де Subject був публічним, рано чи пізно з'являлись баги через те, що два компоненти викликали .next() незалежно один від одного.

ReplaySubject

ReplaySubject(n) буферизує останні n значень. Пізній підписник отримує їх одразу при підписці, а потім продовжує отримувати нові.

typescript

import { ReplaySubject } from 'rxjs';

const chat$ = new ReplaySubject<string>(2); // зберігати останні 2 повідомлення

chat$.next('Привіт');
chat$.next('Як справи?');
chat$.next('Ти тут?'); // буфер тепер ['Як справи?', 'Ти тут?']

// Пізній підписник (користувач відкрив панель чату)
chat$.subscribe(msg => console.log('Відтворено:', msg));
// → Відтворено: Як справи?
// → Відтворено: Ти тут?

Без параметра розміру new ReplaySubject() тримає кожне значення в пам'яті весь час роботи застосунку. В довгоживучих застосунках це витік пам'яті. Використовуй new ReplaySubject(100, 5000), щоб обмежити буфер до 100 значень за останні 5 секунд.

AsyncSubject

AsyncSubject збирає всі значення, але нічого не видає до виклику .complete(). Після завершення він відправляє тільки останнє значення всім поточним і майбутнім підписникам.

typescript

import { AsyncSubject } from 'rxjs';

const result$ = new AsyncSubject<number>();

result$.subscribe(v => console.log('Результат:', v));

result$.next(1);
result$.next(2);
result$.next(3);

result$.complete();
// → Результат: 3

В Angular UI застосовується рідко. Добре підходить для обгортання одноразових операцій, де важливий лише фінальний результат, схоже до Promise.

Коли що використовувати

Кліки, WebSocket-повідомлення, DOM-події: звичайний Subject (без потреби в історії)
Статус авторизації, обрана тема, поточний користувач: BehaviorSubject (кожен підписник потребує актуального стану)
Останні повідомлення чату, стек скасувань, N останніх показань датчика: ReplaySubject(n)
Одноразовий HTTP-запит: AsyncSubject (або просто Promise)

Типові помилки

1. Очікування минулих значень від звичайного Subject

typescript

const subject = new Subject<string>();
subject.next('Пропущено'); // emit до появи будь-якого підписника

subject.subscribe(v => console.log(v)); // нічого не отримає

Звичайний Subject скидає значення, якщо на момент emit не було підписників. Використовуй BehaviorSubject('default') або ReplaySubject(10), коли пізні підписники повинні отримати попередні значення.

2. Забування про відписку та завершення

typescript

const subject = new Subject<string>();
subject.subscribe(() => {}); // нічого не відписується
// subject.complete() також ніколи не викликається
// підписник залишається в пам'яті до закриття вкладки

Викликай subject.complete() коли Subject більше не потрібен, або використовуй takeUntil(this.destroy$) в Angular-компонентах. В шаблонах async pipe прибирає підписки автоматично.

3. Виклик .next() після .complete()

typescript

const subject = new Subject<string>();
subject.complete();
subject.next('Запізно'); // тихо ігнорується в RxJS v7+

Після завершення Subject перебуває в термінальному стані. Будь-який .next() після цього не спрацює. Якщо потрібен новий потік, створи новий Subject.

4. Публічний Subject без asObservable()

typescript

// Уникай такого патерну в Angular-сервісах
public userSubject = new BehaviorSubject<User | null>(null);
// Будь-який споживач може викликати .next() ззовні

Оголоси Subject private, а назовні відкривай тільки .asObservable(). Це конвенція, а не жорстка помилка, але вона захищає від неузгоджених мутацій стану.

5. ReplaySubject без обмеження буфера

typescript

// Витік пам'яті в довгоживучому застосунку
const history$ = new ReplaySubject<AppState>(); // без ліміту
// росте безкінечно разом з кожним emit

Завжди передавай розмір буфера. Додавай часове вікно другим аргументом, якщо дані мають природний термін придатності.

Де зустрічається в реальних проєктах

Angular-сервіси: BehaviorSubject для сесії користувача, feature flags, активних параметрів маршруту
NgRx: потоки дій (action streams) всередині effects використовують Subjects
NestJS: WebSocket-шлюзи транслюють повідомлення клієнтам через Subjects
React з rxjs-hooks: реактивний стан без Redux-шаблонного коду
socket.io: обгортання event emitter у Subject для чистого Observable API

Питання для поглиблення

Q: Яка різниця між Subject і shareReplay() на звичайному Observable?
A: shareReplay() бере холодний Observable і робить його гарячим з буфером відтворення. Subject дає ручний контроль над тим, коли передаються значення. Використовуй shareReplay() коли є наявне джерело Observable, а Subject коли потрібно передавати значення вручну.

Q: Чому BehaviorSubject безпечно використовувати в Angular з zone.js?
A: Він emit синхронно при підписці, а це відбувається всередині Angular zone. Zone.js виявляє emit і планує визначення змін. async pipe у шаблонах підписується всередині зони автоматично.

Q: Як розмір буфера ReplaySubject впливає на пам'ять?
A: Фіксований розмір на зразок ReplaySubject(10) витісняє старі значення при надходженні нових, тримаючи пам'ять стабільною. ReplaySubject() без аргументів росте безмежно. Комбінуй розмір і часове вікно: new ReplaySubject(100, 5000) зберігає щонайбільше 100 значень за останні 5 секунд.

Q: Чи може Subject бути холодним?
A: Ні. Всі Subject в RxJS є гарячими за своєю природою. Вони не створюють нового виконання на кожного підписника. Для холодної поведінки використовуй defer() або new Observable() напряму.

Q: Що відбувається з Subject під час server-side rendering в Angular?
A: Emit виконується і на сервері. Якщо підписник оновлює стан компонента, це може спричинити невідповідність при гідратації на клієнті. Захищай серверні emit через isPlatformBrowser(platformId) з inject(PLATFORM_ID).

Приклади

Звичайний Subject: шина подій для кількох обробників

typescript

import { Subject } from 'rxjs';

const click$ = new Subject<MouseEvent>();

// Один слухач подій живить кілька обробників
document.getElementById('btn')!.addEventListener('click', e => click$.next(e));

click$.subscribe(e => console.log('Обробник 1: клік на', e.clientX));
click$.subscribe(e => analytics.track('button_click', { x: e.clientX }));

// Обидва обробники спрацьовують на кожен клік
// Не потрібно додавати окремий addEventListener для кожного обробника

Одне джерело події обслуговує кількох споживачів без додаткових DOM-слухачів. При знищенні компонента один виклик click$.complete() прибирає всіх підписників.

BehaviorSubject: стан авторизації в Angular з патерном private/public

typescript

import { Injectable } from '@angular/core';
import { BehaviorSubject } from 'rxjs';
import { User } from './user.model';

@Injectable({ providedIn: 'root' })
export class AuthService {
  private userSubject = new BehaviorSubject<User | null>(null);
  public user$ = this.userSubject.asObservable();

  login(user: User) {
    this.userSubject.next(user);
  }

  logout() {
    this.userSubject.next(null);
  }
}

// Компонент header: завжди має актуальний стан авторизації
authService.user$.subscribe(user => {
  this.isLoggedIn = user !== null;
  // null одразу при завантаженні, User-об'єкт після логіну
});

Оскільки BehaviorSubject emit синхронно при підписці, заголовок рендериться з правильним станом авторизації з першого кадру. Початкове значення null означає "не авторизований" без жодного додаткового стану завантаження.

ReplaySubject: стек скасувань з обмеженням розміру і часу

typescript

import { ReplaySubject } from 'rxjs';

interface EditorState { content: string; cursor: number; }

// Зберігати останні 10 станів, відкидати старші 30 секунд
const undoHistory$ = new ReplaySubject<EditorState>(10, 30000);

function saveState(state: EditorState) {
  undoHistory$.next(state);
}

function openUndoPanel() {
  // Підписник одразу отримує до 10 останніх станів
  undoHistory$.subscribe(state => renderUndoItem(state));
}

saveState({ content: 'Привіт', cursor: 6 });
saveState({ content: 'Привіт світ', cursor: 11 });

openUndoPanel();
// → renderUndoItem викликається для кожного збереженого стану
// Стани старші 30с автоматично видаляються

Часове вікно (30 000 мс) означає, що стани з сесії, яка була неактивна 30 секунд, не з'являться в панелі скасувань. Це відповідає очікуванням користувача краще, ніж показувати правки годинної давності.

Коротка відповідь

Для співбесіди

Premium

Коротка відповідь допоможе вам впевнено відповідати на цю тему під час співбесіди.

Дочитали статтю?