Nền tảng

Khi một màn hình vừa nhận DOM event, vừa gọi API, vừa nghe WebSocket và còn phải hủy công việc cũ, phần khó không nằm ở từng callback. Phần khó là trả lời nhất quán: dữ liệu nào đến trước, lỗi đi đâu, khi nào luồng kết thúc và ai chịu trách nhiệm dọn tài nguyên.

Nhóm Nền tảng giúp bạn xây mental model để trả lời những câu hỏi đó trước khi học hàng chục operator. Đây là trang định hướng: bạn sẽ chạy một pipeline nhỏ, biết cách đọc nó và chọn đúng bài chi tiết để học tiếp.

Mục lụcCopy Anchor Link

• Nền tảng giúp bạn làm được gì
• Mental model của RxJS
  • Luồng dữ liệu đi xuôi
  • Quyền dừng đi ngược
• Đọc một pipeline từ đầu đến cuối
  • Code TypeScript
  • Kết quả và vòng đời
• Bản đồ năm bài nền tảng
• Cách đi qua nhóm nền tảng
  • Nếu bạn mới bắt đầu
  • Nếu bạn đã từng dùng RxJS
• Checklist để đọc một stream
• Chọn RxJS hay công cụ đơn giản hơn
• Những bẫy nên nhận ra sớm
• Đi tiếp từ nền tảng
• Nguồn tham khảo

Nền tảng giúp bạn làm được gìCopy Anchor Link

Sau nhóm này, mục tiêu không phải là nhớ thật nhiều tên operator. Bạn cần nhìn một đoạn RxJS và chỉ ra được:

• Producer nào đang tạo dữ liệu: mảng, timer, DOM event, request hay socket.
• Observable mô tả execution nào và execution đó bắt đầu lúc nào.
• Operator đang biến đổi giá trị hay thay đổi quy tắc về thời gian, lỗi và concurrency.
• Observer nhận next, error, complete ở đâu.
• Subscription thuộc về ai và teardown chạy khi nào.

Nếu chưa trả lời được năm câu này, thêm operator thường chỉ làm pipeline khó đoán hơn. Nếu đã trả lời được, những chủ đề như multicasting, switchMap, retry hay scheduler sẽ có chỗ rõ ràng trong cùng một mô hình.

Mental model của RxJSCopy Anchor Link

Hãy hình dung RxJS như một dây chuyền đang hoạt động, không phải một mảng nằm sẵn trong bộ nhớ. Producer đưa giá trị vào; Observable cùng các operator mô tả đường đi; Observer nhận notification; Subscription giữ quyền dừng execution.

Producer          Observable và operators                 Observer
DOM event  ───►  source$ ─► map ─► filter ────────────►  next(value)
timer      ───►                                         error(error)
WebSocket  ───►                                         complete()
                       ▲                                      │
                       └──── teardown ◄── Subscription ◄──────┘

Sơ đồ dùng ASCII vì project hiện chỉ đăng ký bộ MDX component mặc định và chưa cấu hình Mermaid.

Luồng dữ liệu đi xuôiCopy Anchor Link

Producer gửi notification theo một contract đơn giản:

• next(value) có thể xuất hiện không lần nào, một lần hoặc nhiều lần.
• error(error) kết thúc execution do lỗi.
• complete() kết thúc execution bình thường.
• error và complete loại trừ nhau; sau terminal notification sẽ không còn next hợp lệ.

Operator nhận Observable và trả về Observable mới. Vì vậy, pipe(...) xây mô tả mới chứ không sửa source tại chỗ. Với phần lớn cold Observable, mô tả ấy còn lazy: chưa có subscribe(), producer chưa bắt đầu chạy.

Đây cũng là điểm mà phép so sánh với mảng dừng lại. Mảng chứa các phần tử đã có; Observable có thể đại diện cho giá trị chưa xuất hiện, kéo dài vô hạn, phát đồng bộ hoặc bất đồng bộ, và cần cleanup khi consumer rời đi.

Quyền dừng đi ngượcCopy Anchor Link

subscribe() nối Observer vào pipeline và trả về một Subscription. Khi consumer gọi unsubscribe(), tín hiệu dừng đi ngược qua chuỗi để source cùng các operator giải phóng timer, listener, request hoặc kết nối mà chúng sở hữu.

Có ba đường kết thúc cần phân biệt:

Điểm dễ nhầm nhất là hàng cuối: unsubscribe không phải complete. Nếu cleanup phải chạy trên mọi đường kết thúc, hãy đặt nó trong teardown của source hoặc dùng finalize, đừng chỉ đặt trong callback complete.

Đọc một pipeline từ đầu đến cuốiCopy Anchor Link

Giả sử một dashboard nhận nhiệt độ từ cảm biến. Ví dụ rút gọn dưới đây phát ba số đo ban đầu rồi giữ phiên kết nối mở cho đến khi consumer chủ động dừng. Source ngoài đời có thể là WebSocket; source đồng bộ ở đây giúp thứ tự lifecycle dễ quan sát.

Code TypeScriptCopy Anchor Link

import { Observable, filter, finalize, map } from 'rxjs';

const temperature$ = new Observable<number>((subscriber) => {
  console.log('[source] kết nối');

  subscriber.next(28);
  subscriber.next(31);
  subscriber.next(35);

  return () => {
    console.log('[source] ngắt kết nối');
  };
});

const alert$ = temperature$.pipe(
  filter((celsius) => celsius >= 30),
  map((celsius) => `Cảnh báo: ${celsius}°C`),
  finalize(() => console.log('[pipeline] finalize')),
);

const subscription = alert$.subscribe({
  next: (message) => console.log(message),
  error: (error: unknown) => console.error('[observer] lỗi', error),
  complete: () => console.log('[observer] complete'),
});

subscription.unsubscribe();
console.log('closed:', subscription.closed);

Output:

[source] kết nối
Cảnh báo: 31°C
Cảnh báo: 35°C
[source] ngắt kết nối
[pipeline] finalize
closed: true

Kết quả và vòng đờiCopy Anchor Link

Có thể đọc ví dụ theo sáu bước:

1. new Observable(...) tạo mô tả source; callback chưa chạy tại thời điểm này.
2. pipe(...) tạo alert$, vẫn chưa kết nối producer.
3. subscribe(...) bắt đầu execution nên [source] kết nối xuất hiện.
4. filter loại 28; map đổi hai số còn lại thành message trước khi Observer nhận chúng.
5. Source không gọi complete(), nên phiên vẫn mở sau ba giá trị và callback complete không chạy.
6. unsubscribe() chạy teardown của source và finalize; sau đó subscription.closed là true.

Ví dụ này cố ý cho thấy cleanup là một phần của contract, không phải việc phụ làm sau. Trong source thực tế, hàm teardown sẽ gọi removeEventListener, clearInterval, WebSocket.close() hoặc AbortController.abort() tùy tài nguyên bên dưới.

Bản đồ năm bài nền tảngCopy Anchor Link

Mỗi bài trả lời một câu hỏi khác nhau. Đọc chúng như các lớp của cùng một mô hình, không phải năm định nghĩa rời rạc.

Trang index chỉ cung cấp bản đồ. Khi cần contract chi tiết, ví dụ edge case hoặc tiêu chí ra quyết định, hãy mở đúng bài thay vì suy ra từ đoạn overview ngắn này.

Cách đi qua nhóm nền tảngCopy Anchor Link

Nếu bạn mới bắt đầuCopy Anchor Link

Đi theo thứ tự trong bảng. Sau mỗi bài, tự lấy một tình huống quen thuộc như ô tìm kiếm hoặc timer và gọi tên producer, Observable, Observer, Subscription. Cách này hữu ích hơn học thuộc định nghĩa vì nó buộc mental model gắn với code.

Sau Vòng đời của stream, hãy quay lại ví dụ cảm biến phía trên và thử trả lời: điều gì thay đổi nếu source gọi complete() ngay sau 35? Bạn nên dự đoán callback complete, teardown, finalize và trạng thái closed trước khi chạy code.

Nếu bạn đã từng dùng RxJSCopy Anchor Link

Bạn có thể đọc theo chỗ hổng thay vì theo thứ tự. Nếu hay quên unsubscribe, bắt đầu ở vòng đời stream. Nếu pipeline đầy nested subscribe hoặc Subject toàn cục, ôn lại Observer pattern và bài chọn RxJS. Nếu vẫn gọi mọi Observable là “async”, xem lại source đồng bộ trong ví dụ phía trên.

Dù đi đường nào, đừng bỏ bài Khi nào dùng RxJS?. Biết viết RxJS và biết lúc không nên viết là hai kỹ năng khác nhau.

Checklist để đọc một streamCopy Anchor Link

Khi gặp một pipeline chưa quen, đọc từ boundary thay vì đoán theo tên biến:

1. Source là gì? Nó phát đồng bộ hay chờ event? Hữu hạn hay có thể sống mãi?
2. Khi nào source bắt đầu? Lúc tạo Observable, lúc subscribe hay đã chạy độc lập từ trước?
3. Mỗi operator đổi điều gì? Giá trị, thời gian, số lượng emission, error path hay số execution?
4. Có bao nhiêu subscription? Mỗi subscription tạo producer riêng hay dùng chung producer?
5. Terminal signal đi đâu? Lỗi được bắt hay làm kết thúc toàn chuỗi? Stream có tự complete không?
6. Ai sở hữu teardown? Lifecycle nào gọi unsubscribe, và tài nguyên bên dưới có thật sự hủy được không?

Đây là checklist nên mang sang mọi phần của tài liệu. Nó ngăn hai lỗi phổ biến: chỉ nhìn happy path của next, và cho rằng mọi Observable đều có cùng hành vi hot, cold, sync hoặc async.

Chọn RxJS hay công cụ đơn giản hơnCopy Anchor Link

Mặc định, mình chọn công cụ ít abstraction nhất nhưng vẫn diễn đạt đúng bài toán:

RxJS không làm complexity biến mất; nó chuyển complexity về một ngôn ngữ gồm Observable, operator và lifecycle. Chỉ nên trả chi phí đó khi pipeline làm quan hệ dữ liệu theo thời gian rõ hơn phương án imperative.

Những bẫy nên nhận ra sớmCopy Anchor Link

• Cho rằng Observable luôn bất đồng bộ. of(1, 2, 3) có thể phát và complete ngay bên trong lời gọi subscribe(); scheduler và source mới quyết định timing.
• Subscribe nhiều lần mà không xét lại producer. Cold Observable thường chạy lại source cho từng subscriber, có thể tạo request, timer hoặc side effect trùng lặp.
• Đặt subscribe bên trong subscribe. Cách này tách error handling và teardown thành nhiều nhánh. Thường nên biểu diễn tác vụ bên trong thành Observable rồi chọn flattening operator phù hợp.
• Dùng Subject như global event bus mặc định. Khi nhiều nơi vừa đọc vừa gọi next, data flow và ownership khó truy vết. Hãy ưu tiên Observable chỉ đọc ở boundary.
• Đồng nhất unsubscribe với complete. Consumer hủy sẽ chạy teardown nhưng không làm callback complete chạy.
• Học operator trước lifecycle. Biết map và switchMap chưa đủ nếu bạn không biết source bắt đầu, kết thúc và bị hủy ở đâu.

Khi gặp các vấn đề này trong code thật, Nested subscribe, Memory leak và Cold và hot Observable là ba điểm tra cứu hữu ích.

Đi tiếp từ nền tảngCopy Anchor Link

Nếu bạn chưa từng viết pipeline, hãy bắt đầu bằng bài tổng quan rồi đi tuần tự. Nếu năm câu hỏi ở checklist đã rõ, chuyển sang nhóm Bắt đầu để cài đặt, đọc marble diagram và debug execution thật.

Nguồn tham khảoCopy Anchor Link

• RxJS — Introduction — tổng quan chính thức về Observable, Observer, Subscription và operators.
• RxJS — Observable — execution, subscription và ví dụ Observable đồng bộ.
• RxJS — Subscription — cơ chế hủy execution và gom teardown.
• RxJS 7.8.2 source — Observable — hành vi subscribe và đăng ký teardown.
• RxJS 7.8.2 source — finalize — callback chạy khi complete, error hoặc unsubscribe.
• Gói rxjs trên npm — thông tin phiên bản phát hành và package TypeScript.