Observer pattern

Một ô tìm kiếm có thể phát ra hàng chục sự kiện khi người dùng gõ, còn component chỉ muốn phản ứng với những giá trị mình quan tâm và dừng nghe khi bị huỷ. Nếu nối trực tiếp mọi callback với mọi nguồn sự kiện, phần khởi động, xử lý lỗi và dọn tài nguyên rất nhanh bị rải khắp code. Observer pattern tách bên phát dữ liệu khỏi bên nhận; RxJS bổ sung Observable và Subscription để mối liên hệ đó có vòng đời rõ ràng và ghép nối được.

Mục lụcCopy Anchor Link

• Mental model: đăng ký một phiên làm việc
• Bốn vai trò cốt lõi
• Điều gì xảy ra khi subscribe
  • Ví dụ đồng bộ tối thiểu
  • Ba notification và một hành động huỷ
• Teardown: dọn tài nguyên đúng chỗ
• Ví dụ thực tế: lắng nghe ô tìm kiếm
• Chọn API nào
• Những lỗi tư duy thường gặp
  • Nghĩ rằng khai báo Observable là đã chạy
  • Nghĩ rằng mọi Observable đều async
  • Subscribe nhiều lần nhưng mong một execution duy nhất
  • Quên teardown trong custom Observable
  • Dùng complete như cleanup duy nhất
  • Subscribe bên trong subscribe
• Bước tiếp theo
• Tài liệu tham khảo

Mental model: đăng ký một phiên làm việcCopy Anchor Link

Hình dung một kênh thông báo ở quán ăn: bếp là producer, quy ước chuyển phiếu là Observable, nhân viên phục vụ nhận phiếu là Observer, còn tờ đăng ký ca trực là Subscription. Khi ca kết thúc, tờ đăng ký bị huỷ và tài nguyên liên quan phải được dọn.

Ẩn dụ này chỉ giúp nhớ vai trò. Trong RxJS, một Observable thường có thể tạo producer riêng cho từng lần subscribe, trong khi bếp ngoài đời thường được nhiều người dùng chung. Việc producer riêng hay dùng chung dẫn tới khái niệm cold, hot, unicast và multicast ở phần sau của lộ trình.

                                next(value)
┌──────────┐   được bao bởi   ┌────────────┐ ─────────────► ┌──────────┐
│ Producer │ ───────────────► │ Observable │   error(err)   │ Observer │
└──────────┘                  └─────┬──────┘ ─────────────► └──────────┘
      ▲                             │          complete()         │
      │                             │                             │
      └──── teardown ◄──────────────┴──── Subscription ◄─────────┘
                                             unsubscribe()

Điểm quan trọng nhất: Observable không phải dữ liệu đã chạy sẵn. Nó là một template mô tả cách nối consumer với producer. Với Observable thông thường, subscribe() mới thiết lập kết nối và bắt đầu execution.

Bốn vai trò cốt lõiCopy Anchor Link

Producer và Observable không nhất thiết là một. Với fromEvent, DOM là producer; Observable là lớp bao cung cấp contract và khả năng kết hợp operator. Với new Observable, code tạo timer nằm trong hàm khởi tạo execution nên ranh giới giữa hai khái niệm trông gần nhau hơn, nhưng vẫn nên hỏi riêng: “Ai sinh dữ liệu?” và “Ai quản lý cách đăng ký?”.

Observer là consumer được biểu diễn bằng object. Một observer đầy đủ có ba handler:

import type { Observer } from 'rxjs';

const observer: Observer<number> = {
  next: (value) => console.log('giá trị:', value),
  error: (error: unknown) => console.error('lỗi:', error),
  complete: () => console.log('hoàn tất'),
};

RxJS cũng chấp nhận partial observer, chẳng hạn chỉ có next. Tuy vậy, ở boundary của ứng dụng như gọi API hoặc kết nối bên ngoài, mình thường khai báo error rõ ràng để lỗi không trở thành unhandled error khó truy vết.

Điều gì xảy ra khi subscribeCopy Anchor Link

Mỗi lời gọi subscribe tạo một subscription. Theo thứ tự khái niệm, RxJS sẽ:

1. nhận observer từ consumer;
2. thiết lập Observable execution và kết nối tới producer;
3. chuyển các notification hợp lệ tới observer;
4. trả về một Subscription để consumer có thể huỷ execution;
5. chạy teardown khi execution kết thúc hoặc bị huỷ.

Ví dụ đồng bộ tối thiểuCopy Anchor Link

Ví dụ sau cố ý dùng new Observable để lộ toàn bộ vòng đời. Trong code ứng dụng, creation function có sẵn thường là lựa chọn gọn và an toàn hơn.

import { Observable, type Observer } from 'rxjs';

const numbers$ = new Observable<number>((subscriber) => {
  console.log('producer: bắt đầu');

  subscriber.next(10);
  subscriber.next(20);
  subscriber.complete();

  return () => console.log('producer: teardown');
});

const observer: Observer<number> = {
  next: (value) => console.log('observer: next', value),
  error: (error: unknown) => console.error('observer: error', error),
  complete: () => console.log('observer: complete'),
};

console.log('trước subscribe');
const subscription = numbers$.subscribe(observer);
console.log('sau subscribe, closed =', subscription.closed);

Kết quả:

trước subscribe
producer: bắt đầu
observer: next 10
observer: next 20
observer: complete
producer: teardown
sau subscribe, closed = true

next xuất hiện trước dòng “sau subscribe”, vì Observable có thể chạy đồng bộ. RxJS không tự biến mọi việc thành async. Sau complete, subscription đóng và teardown chạy; vì thế subscription.closed là true.

Đừng gọi biến được truyền vào constructor là observer của ứng dụng. Đó là Subscriber do RxJS cung cấp cho phía producer: nó chuyển tiếp notification tới observer và bảo vệ Observable contract. Consumer vẫn chỉ truyền observer vào subscribe.

Ba notification và một hành động huỷCopy Anchor Link

Một execution tuân theo grammar ngắn gọn:

next* (error | complete)?

Nói cách khác, nó có thể phát không, một hoặc nhiều next; sau đó có nhiều nhất một terminal notification là error hoặc complete. Khi một terminal notification đã xảy ra, observer không nhận thêm giá trị.

• next(value) mang dữ liệu và có thể xảy ra nhiều lần.
• error(error) báo execution thất bại rồi đóng subscription.
• complete() báo producer kết thúc bình thường rồi đóng subscription.
• unsubscribe() là hành động của consumer để ngừng quan sát; nó không phải notification.

Điểm cuối thường gây nhầm: gọi unsubscribe() không gọi handler complete. complete là tín hiệu từ producer, còn unsubscribe là consumer nói “tôi không cần nữa”. Cả hai đều dẫn tới finalization, nhưng ý nghĩa nghiệp vụ khác nhau.

Để đi sâu vào quy tắc terminal notification, đọc Vòng đời của stream.

Teardown: dọn tài nguyên đúng chỗCopy Anchor Link

Producer có thể giữ timer, event listener, socket hoặc request đang chạy. Khi tạo custom Observable, hàm subscribe nên trả về teardown để giải phóng tài nguyên đó. RxJS chạy teardown khi subscription complete, error hoặc bị unsubscribe.

import { Observable } from 'rxjs';

const heartbeat$ = new Observable<number>((subscriber) => {
  let beat = 0;

  const timerId = setInterval(() => {
    beat += 1;
    subscriber.next(beat);
  }, 1_000);

  return () => {
    clearInterval(timerId);
    console.log('teardown: đã đóng timer');
  };
});

const subscription = heartbeat$.subscribe({
  next: (beat) => console.log('heartbeat', beat),
  complete: () => console.log('hoàn tất'),
});

setTimeout(() => subscription.unsubscribe(), 2_500);

Kết quả xấp xỉ:

heartbeat 1
heartbeat 2
teardown: đã đóng timer

Dòng hoàn tất không xuất hiện vì consumer đã unsubscribe chứ producer không gọi complete(). Nếu bỏ clearInterval, observer đã đóng sẽ không nhận thêm giá trị, nhưng timer vẫn đánh thức event loop và lãng phí tài nguyên. Đây là lý do “không còn thấy log” chưa chứng minh việc cleanup đã đúng.

Creation function và operator của RxJS thường đã cài teardown đúng. Chỉ dùng new Observable khi bạn thực sự cần bọc một API callback hoặc resource chưa có adapter phù hợp. Xem cách tổ chức cleanup phức tạp hơn tại Subscription và teardown và Tự tạo Observable.

Ví dụ thực tế: lắng nghe ô tìm kiếmCopy Anchor Link

Giả sử trang có một ô nhập với id="search". Bạn muốn chỉ xử lý khi người dùng ngừng gõ 300 ms và bỏ qua query trùng liên tiếp:

import {
  debounceTime,
  distinctUntilChanged,
  fromEvent,
  map,
} from 'rxjs';

const input = document.querySelector<HTMLInputElement>('#search');

if (!input) {
  throw new Error('Không tìm thấy #search');
}

const query$ = fromEvent<InputEvent>(input, 'input').pipe(
  map((event) =>
    (event.currentTarget as HTMLInputElement).value.trim(),
  ),
  debounceTime(300),
  distinctUntilChanged(),
);

const subscription = query$.subscribe({
  next: (query) => console.log('tìm kiếm:', query),
  error: (error: unknown) => console.error('ô tìm kiếm lỗi:', error),
});

export function destroySearchBox(): void {
  subscription.unsubscribe();
}

Nếu người dùng gõ nhanh r → rx → rxjs, rồi dừng hơn 300 ms, output là:

tìm kiếm: rxjs

Ở đây:

• DOM input event là producer;
• query$ là Observable mô tả pipeline;
• object truyền vào subscribe là observer;
• subscription quản lý lần lắng nghe cụ thể;
• fromEvent đăng ký event listener khi subscribe và gỡ listener trong teardown khi destroySearchBox() chạy.

DOM event không tự complete khi element biến mất. Vì vậy lifecycle của UI phải gọi destroySearchBox; chỉ xoá element khỏi DOM không thay thế cho unsubscribe. Khi nối query với HTTP, nên dùng flattening operator thay vì subscribe lồng nhau; xem Nested subscribe.

Chọn API nàoCopy Anchor Link

Trong RxJS 7.x, các overload subscribe(next, error, complete) nhiều đối số đã bị deprecated. Đừng viết callback rời theo vị trí; observer object dễ đọc hơn và tránh phải truyền null làm chỗ trống.

Một câu hỏi khác là khi nào cần lưu Subscription. Với source hữu hạn, đồng bộ như of(1, 2, 3), execution đã đóng ngay sau subscribe. Với timer, DOM event, WebSocket hoặc source sống lâu, bạn cần chiến lược huỷ rõ ràng: giữ Subscription, gắn nó vào lifecycle của framework, hoặc dùng operator kết thúc thích hợp. Không nên gọi unsubscribe() ngay sau subscribe() chỉ như một nghi thức; làm vậy có thể huỷ công việc async trước khi nó phát giá trị.

Những lỗi tư duy thường gặpCopy Anchor Link

Nghĩ rằng khai báo Observable là đã chạyCopy Anchor Link

Observable thường lazy. Tạo const data$ = new Observable(...) chỉ tạo template; side effect trong hàm subscribe chưa chạy cho tới khi có consumer đăng ký.

Nghĩ rằng mọi Observable đều asyncCopy Anchor Link

Ví dụ of(1, 2, 3) phát và complete đồng bộ theo mặc định. Nếu thứ tự thực thi quan trọng, đừng đoán dựa vào kiểu Observable; hãy xem producer và scheduler thực tế.

Subscribe nhiều lần nhưng mong một execution duy nhấtCopy Anchor Link

Với cold Observable, mỗi subscribe thường tạo producer riêng. Hai subscription vào một HTTP Observable có thể gây hai request. Nếu cần chia sẻ một producer, hãy học rõ trade-off của multicast thay vì thêm cờ thủ công; bắt đầu tại Cold và hot Observable và Unicast và multicast.

Quên teardown trong custom ObservableCopy Anchor Link

Chặn notification sau unsubscribe chưa đủ; timer, listener và socket vẫn phải được đóng. Mỗi lệnh setup giữ resource nên có hành động cleanup đối xứng trong teardown.

Dùng complete như cleanup duy nhấtCopy Anchor Link

Source sống lâu có thể không complete, còn unsubscribe không gọi callback complete. Đặt cleanup resource ở teardown hoặc dùng operator finalize cho side effect cần chạy khi subscription kết thúc theo bất kỳ đường nào.

Subscribe bên trong subscribeCopy Anchor Link

Nested subscribe tách một flow thành nhiều subscription có ownership và error handling rời nhau. Trừ khi bạn chủ ý quản lý từng execution độc lập, hãy ưu tiên operator kết hợp hoặc flattening để teardown lan truyền qua một subscription chính.

Bước tiếp theoCopy Anchor Link

Bạn có thể tự kiểm tra mental model bằng cách lấy ví dụ ô tìm kiếm và trả lời bốn câu: producer là gì, Observable được tạo ở đâu, observer xử lý notification nào, và ai chịu trách nhiệm unsubscribe. Nếu một câu chưa rõ, lifecycle trong code cũng thường chưa rõ.

Tài liệu tham khảoCopy Anchor Link

• RxJS 7.8.2 — Observable guide
• RxJS 7.8.2 — Observer guide
• RxJS 7.8.2 — Subscription guide
• RxJS 7.8.2 — Glossary and semantics