RxJS trong Angular 2026: Operator, Subject và Interop Signals

RxJS trong Angular vẫn là xương sống của lập trình bất đồng bộ vào năm 2026, ngay cả khi Signals định hình lại cách các component Angular theo dõi state đồng bộ. Biết khi nào nên dùng Observable, khi nào Signal phù hợp hơn, và hai mô hình này phối hợp với nhau ra sao giờ đây là kỹ năng cốt lõi của bất kỳ lập trình viên Angular nào. Bài phân tích chuyên sâu này đi qua các operator, Subject và những mẫu interop tạo nên sự khác biệt giữa một câu trả lời phỏng vấn tự tin và một câu trả lời bị khựng lại.

Vì sao RxJS vẫn quan trọng song hành cùng Angular Signals

Sự xuất hiện của Signals khiến nhiều nhóm tự hỏi liệu RxJS có đang trên đường bị loại bỏ hay không. Câu trả lời, sau ba phiên bản lớn, là không. Signals xuất sắc trong việc biểu diễn state có giá trị ngay lúc này: một bộ đếm, một trường biểu mẫu, một tab được chọn. Observables xuất sắc trong việc mô hình hóa các sự kiện đến theo thời gian và có thể cần biến đổi, hủy bỏ hoặc kết hợp: phản hồi HTTP, tin nhắn WebSocket, input bàn phím, sự kiện router và bộ đếm thời gian.

Chính các API của Angular làm cho sự phân chia này lộ rõ. HttpClient vẫn trả về Observables. Reactive forms phơi bày valueChanges và statusChanges dưới dạng Observables. Các sự kiện router chảy qua một Observable. Bất cứ nơi nào nhiều sự kiện bất đồng bộ đi qua cùng một kênh, các operator RxJS vẫn là công cụ giàu tính biểu đạt nhất hiện có. Hướng dẫn reactivity của Angular chính thức xem hai mô hình này là bổ trợ cho nhau chứ không cạnh tranh, và tài liệu RxJS vẫn là tài liệu tham khảo bắt buộc cho bất kỳ luồng dữ liệu không tầm thường nào.

Quy tắc thực tiễn đứng vững trong code review: dùng Signals cho state cục bộ của component được đọc trong template, và dùng Observables cho các pipeline bất đồng bộ cần đến operator. Lớp interop kết nối cả hai để không mô hình nào trở thành bức tường ngăn cách.

Các operator kết hợp củng cố thêm luận điểm này. Phối hợp nhiều nguồn bất đồng bộ, chẳng hạn hợp nhất tham số route với bộ lọc đã lưu của người dùng và một luồng giá trực tiếp, chính là điều mà combineLatest, withLatestFrom và forkJoin được tạo ra để làm. Signals có thể suy ra giá trị được tính toán từ các signal khác, nhưng không thể diễn đạt sự phối hợp dựa trên thời gian như "chỉ phát ra sau khi cả ba nguồn đã tạo ra ít nhất một giá trị." Sự kết hợp khai báo theo thời gian đó là lý do bền bỉ khiến RxJS vẫn nằm trong bộ công cụ.

Các operator RxJS cốt lõi mà mọi lập trình viên Angular dựa vào

Operator là các hàm thuần khiết nhận vào một Observable và trả về một Observable mới. Kịch bản phỏng vấn kinh điển là search-as-you-type, vì nó rèn luyện bốn operator cùng lúc: điều tiết input, bỏ qua trùng lặp, hủy các yêu cầu đã cũ và phục hồi khỏi lỗi.

import { Component, inject } from '@angular/core';
import { FormControl, ReactiveFormsModule } from '@angular/forms';
import { HttpClient } from '@angular/common/http';
import { debounceTime, distinctUntilChanged, switchMap, catchError } from 'rxjs/operators';
import { of } from 'rxjs';

@Component({
  selector: 'app-search',
  imports: [ReactiveFormsModule],
  template: `<input [formControl]="query" placeholder="Search repositories" />`,
})
export class SearchComponent {
  private http = inject(HttpClient);
  query = new FormControl('');

  results$ = this.query.valueChanges.pipe(
    debounceTime(300),               // wait until typing pauses for 300ms
    distinctUntilChanged(),          // ignore emissions that repeat the last value
    switchMap((term) =>              // cancel the in-flight request, start a fresh one
      this.http.get(`/api/search?q=${term}`).pipe(
        catchError(() => of([]))     // recover from a failed request without killing the stream
      )
    )
  );
}

Chi tiết tinh tế mà người phỏng vấn dò xét là vì sao switchMap nằm ở lớp ngoài trong khi catchError lồng bên trong nó. Đặt catchError trên Observable HTTP bên trong nghĩa là một yêu cầu thất bại trả về mảng rỗng và luồng bên ngoài vẫn tiếp tục chảy. Chuyển catchError ra pipe bên ngoài sẽ chấm dứt toàn bộ việc tìm kiếm sau lỗi đầu tiên, phá vỡ mọi lần gõ phím tiếp theo. Bốn operator làm phẳng mỗi cái trả lời một câu hỏi về tính đồng thời khác nhau, và chọn sai một cái là một trong những nguyên nhân phổ biến nhất gây race condition trong ứng dụng Angular.

Chọn đúng lựa chọn này chính là điều biến một tính năng chập chờn thành một tính năng có thể dự đoán. Module operator RxJS của SharpSkill rèn giũa những khác biệt này bằng các câu hỏi có chấm điểm tái hiện chính xác các race condition mà mỗi operator ngăn chặn.

Subject và multicasting: BehaviorSubject, ReplaySubject và Subject

Một Observable thông thường là unicast: mỗi subscriber kích hoạt một lần thực thi mới. Một Subject vừa là Observable vừa là Observer, điều này khiến nó multicast. Đặc tính đó biến Subject thành cách chuẩn để xây dựng một state store nhẹ hoặc một event bus bên trong một service.

Ba biến thể khác nhau ở chỗ chúng phát lại điều gì cho các subscriber đến muộn. Một Subject thông thường không phát lại bất cứ điều gì từ quá khứ, nên subscriber chỉ thấy các giá trị đến sau khi nó đăng ký. Một BehaviorSubject lưu giá trị hiện tại và phát lại ngay lập tức, đó là lý do nó cần một giá trị khởi tạo. Một ReplaySubject đệm một số lượng phát ra trong quá khứ có thể cấu hình và phát lại tất cả chúng.

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

@Injectable({ providedIn: 'root' })
export class CartService {
  // BehaviorSubject holds the latest value and replays it to new subscribers
  private readonly itemsSubject = new BehaviorSubject<string[]>([]);

  // expose a read-only stream, never the writable Subject itself
  readonly items$ = this.itemsSubject.asObservable();

  add(item: string): void {
    const current = this.itemsSubject.value;    // synchronous access to the last value
    this.itemsSubject.next([...current, item]); // emit the next immutable state
  }
}

Hai thói quen quan trọng ở đây. Thứ nhất, phơi bày asObservable() thay vì chính Subject ngăn code bên ngoài gọi next() và thay đổi state qua cửa sau. Thứ hai, phát ra một tham chiếu mảng mới thay vì đột biến tại chỗ giữ cho change detection và các phép kiểm tra bằng nhau luôn dự đoán được. Mẫu này có trước Signals và vẫn hoạt động tốt cho state dùng chung, có thể inject được mà nhiều component không liên quan cùng tiêu thụ.

Dù vậy, một BehaviorSubject chỉ dùng thuần để giữ một giá trị hiện tại đơn lẻ được đọc trong template giờ đây thường được biểu diễn tốt hơn dưới dạng một Signal có thể ghi, giúp loại bỏ đăng ký và nghi thức asObservable(). Subject vẫn là lựa chọn mạnh hơn khi store còn cần biến đổi, debounce hoặc kết hợp các lần phát của nó với các luồng khác, vì các operator đó không có tương đương Signal trực tiếp. Quyết định giữa hai bên theo từng trường hợp sử dụng, thay vì mặc định dùng một bên ở mọi nơi, là dấu hiệu của sự thành thạo mà người phỏng vấn tưởng thưởng.

Interop RxJS với Signals bằng toSignal và toObservable

Gói interop cung cấp hai hàm chủ chốt. toSignal chuyển một Observable thành một Signal chỉ đọc mà template có thể tiêu thụ trực tiếp, và nó tự quản lý vòng đời đăng ký. toObservable làm điều ngược lại, biến một Signal thành một Observable để các operator RxJS có thể xử lý những thay đổi của nó.

import { Component, signal, inject } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { toSignal, toObservable } from '@angular/core/rxjs-interop';
import { switchMap } from 'rxjs/operators';

@Component({ selector: 'app-dashboard', template: `{{ user()?.name }}` })
export class DashboardComponent {
  private http = inject(HttpClient);

  // a Signal drives the query; toObservable turns each change into a stream event
  readonly userId = signal(1);
  private readonly user$ = toObservable(this.userId).pipe(
    switchMap((id) => this.http.get<{ name: string }>(`/api/users/${id}`))
  );

  // toSignal subscribes immediately and unsubscribes on destroy, no manual teardown
  readonly user = toSignal(this.user$, { initialValue: null });
}

Tùy chọn initialValue đáng được lưu ý. Nếu không có nó, toSignal trả về một Signal có kiểu là giá trị hoặc undefined, vì chưa có gì được phát ra. Truyền initialValue: null cho template một state khởi đầu được xác định và một kiểu sạch hơn. Với các nguồn đồng bộ như BehaviorSubject, requireSync: true bảo đảm một giá trị luôn có sẵn ngay ở lần đọc đầu tiên và loại bỏ hoàn toàn undefined khỏi kiểu. Cầu nối hai chiều này chính là điều cho phép một codebase áp dụng Signals theo từng bước mà không phải viết lại mọi pipeline Observable hiện có. Module phỏng vấn Angular Signals đồng hành trình bày mô hình reactivity một cách sâu sắc, và tổng quan tính năng Angular 18 Signals lần theo cách các primitive này tiến hóa.

Bắc cầu từ Observables sang signal resource với rxResource

Angular 20 đề cao rxResource như cách khai báo để tải dữ liệu bất đồng bộ từ một Observable vào state dựa trên signal. Nó theo dõi trạng thái loading, error và resolved dưới dạng signal, loại bỏ phần lớn công việc kế toán đăng ký thủ công mà các cuộc gọi HTTP từng đòi hỏi.

import { Component, signal, inject } from '@angular/core';
import { rxResource } from '@angular/core/rxjs-interop';
import { HttpClient } from '@angular/common/http';

@Component({ selector: 'app-products', template: `` })
export class ProductsComponent {
  private http = inject(HttpClient);
  readonly category = signal('laptops');

  // Angular 20 renamed the fields: request -> params and loader -> stream
  readonly products = rxResource({
    params: () => ({ category: this.category() }),
    stream: ({ params }) =>
      this.http.get<Product[]>(`/api/products?category=${params.category}`),
  });
  // products.value(), products.isLoading() and products.error() are all signals
}

Việc đổi tên từ request/loader trong Angular 19 thành params/stream trong Angular 20 là một điểm hay vấp phải, nên nhắc đến tên field hiện tại thể hiện kiến thức cập nhật trong buổi phỏng vấn. Khi signal category thay đổi, rxResource hủy yêu cầu trước và bắt đầu một luồng mới, cho bạn ngữ nghĩa của switchMap một cách miễn phí. Tài liệu API rxResource ghi lại toàn bộ vòng đời trạng thái.

Tránh rò rỉ bộ nhớ với takeUntilDestroyed

Bất kỳ subscribe() thủ công nào sống lâu hơn component của nó đều rò rỉ bộ nhớ và có thể kích hoạt callback lên một view đã bị hủy. Cách khắc phục trong lịch sử là mẫu takeUntil(this.destroy$) với một Subject được hoàn tất trong ngOnDestroy. Angular đã thay thế boilerplate đó bằng takeUntilDestroyed, một operator gắn đăng ký với injection context hiện tại hoặc một DestroyRef tường minh.

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

@Component({ selector: 'app-live-feed', template: `` })
export class LiveFeedComponent {
  private destroyRef = inject(DestroyRef);

  constructor() {
    interval(1000)
      // completes the subscription automatically when the component is destroyed
      .pipe(takeUntilDestroyed(this.destroyRef))
      .subscribe((tick) => console.log('tick', tick));
  }
}

Được gọi bên trong một constructor hoặc một khởi tạo field, takeUntilDestroyed() không cần đối số vì nó đọc DestroyRef sẵn có trong ngữ cảnh. Được gọi ở bất kỳ nơi nào khác, chẳng hạn bên trong một phương thức, nó đòi hỏi một DestroyRef tường minh được truyền vào, như minh họa ở trên. Ưu tiên toSignal và rxResource tránh được đăng ký thủ công hoàn toàn, đây là cách sạch nhất để né tránh rò rỉ; khi thật sự cần một đăng ký thô, takeUntilDestroyed là biện pháp bảo vệ đúng đắn.

Những câu hỏi phỏng vấn RxJS Angular đáng luyện tập

Người phỏng vấn thường xoay quanh cùng những chủ đề nhiều tín hiệu. Trả lời được những câu này một cách gọn gàng phân biệt ứng viên đã từng dùng RxJS với những người chỉ mới đọc về nó.

• Hot so với cold Observables. Một cold Observable khởi động producer của nó theo từng đăng ký, nên mỗi subscriber nhận một lần thực thi độc lập (một yêu cầu HTTP mới). Một hot Observable chia sẻ một producer duy nhất cho mọi subscriber, đó chính là điều mà Subject và share() tạo ra.
• switchMap so với mergeMap so với concatMap so với exhaustMap. switchMap hủy Observable bên trong trước đó (lý tưởng cho tìm kiếm). mergeMap chạy mọi Observable bên trong đồng thời. concatMap xếp hàng chúng theo thứ tự. exhaustMap bỏ qua input mới trong khi một cái đang chạy (lý tưởng để ngăn submit biểu mẫu hai lần).
• Cách ngăn rò rỉ bộ nhớ. Ưu tiên toSignal, rxResource hoặc pipe async, tất cả đều tự hủy đăng ký; với đăng ký thủ công, hãy dùng takeUntilDestroyed.
• Khi nào chuyển một Observable thành một Signal. Chọn toSignal khi template cần một giá trị một cách đồng bộ còn nguồn thì bất đồng bộ, cho phép change detection đọc giá trị mới nhất mà không cần pipe async.

Làm qua các phiên bản có chấm điểm của những câu này dưới áp lực thời gian là cách nhanh nhất để nội hóa chúng. Trung tâm công nghệ Angular nhóm các module RxJS, Signals và change detection ánh xạ trực tiếp tới những câu hỏi này.

Kết luận

RxJS và Signals là đối tác trong stack Angular hiện đại, không phải đối thủ. Những điều đáng mang theo vào dự án hoặc buổi phỏng vấn tiếp theo:

• Dùng Signals cho state đồng bộ của component và Observables cho các pipeline sự kiện bất đồng bộ cần đến operator.
• Chọn switchMap để hủy công việc đã cũ, exhaustMap để chặn trùng lặp, và giữ catchError trên Observable bên trong để luồng bên ngoài sống sót qua các lỗi.
• Phơi bày state của Subject qua asObservable() và phát ra các giá trị bất biến để giữ change detection có thể dự đoán.
• Bắc cầu hai mô hình bằng toSignal và toObservable, và truyền initialValue hoặc requireSync để có kiểu sạch.
• Tải dữ liệu bất đồng bộ theo lối khai báo bằng rxResource trong Angular 20, nhớ các tên field params và stream.
• Bảo vệ mọi đăng ký thủ công còn lại bằng takeUntilDestroyed để loại bỏ rò rỉ.