RxJS ใน Angular 2026: โอเปอเรเตอร์ Subject และ interop Signals

RxJS ใน Angular ยังคงเป็นกระดูกสันหลังของการเขียนโปรแกรมแบบอะซิงโครนัสในปี 2026 แม้ว่า Signals จะเปลี่ยนวิธีที่คอมโพเนนต์ Angular ติดตาม state แบบซิงโครนัสไปแล้วก็ตาม การรู้ว่าเมื่อใดควรเลือกใช้ Observable เมื่อใด Signal เหมาะสมกว่า และทั้งสองทำงานร่วมกันอย่างไร กลายเป็นทักษะหลักของนักพัฒนา Angular ทุกคน บทวิเคราะห์เชิงลึกนี้พาผู้อ่านผ่านโอเปอเรเตอร์ Subject และรูปแบบ interop ที่แยกคำตอบสัมภาษณ์อันมั่นใจออกจากคำตอบที่ติดขัด

เหตุใด RxJS จึงยังสำคัญควบคู่ไปกับ Angular Signals

การเปิดตัว Signals ทำให้หลายทีมตั้งคำถามว่า RxJS กำลังจะถูกเลิกใช้หรือไม่ คำตอบหลังผ่านไปสามเวอร์ชันหลักคือ ไม่ Signals โดดเด่นในการแทน state ที่มีค่าอยู่ในตอนนี้ ได้แก่ ตัวนับ ฟิลด์ในฟอร์ม หรือแท็บที่ถูกเลือก ส่วน Observables โดดเด่นในการจำลองเหตุการณ์ที่เข้ามาตลอดช่วงเวลาและอาจต้องแปลง ยกเลิก หรือรวมเข้าด้วยกัน เช่น การตอบกลับ HTTP ข้อความ WebSocket อินพุตจากคีย์บอร์ด เหตุการณ์ของ router และตัวจับเวลา

API ของ Angular เองก็ทำให้การแบ่งแยกนี้ปรากฏชัด HttpClient ยังคงคืนค่าเป็น Observables reactive forms เปิดเผย valueChanges และ statusChanges เป็น Observables เหตุการณ์ของ router ไหลผ่าน Observable ทุกที่ที่เหตุการณ์อะซิงโครนัสหลายเหตุการณ์ไหลผ่านช่องทางเดียวกัน โอเปอเรเตอร์ RxJS ยังคงเป็นเครื่องมือที่สื่อความหมายได้ดีที่สุดที่มีอยู่ คู่มือ reactivity ของ Angular อย่างเป็นทางการมองว่าทั้งสองโมเดลเป็นส่วนเสริมซึ่งกันและกันมากกว่าจะแข่งขันกัน และ เอกสาร RxJS ยังคงเป็นแหล่งอ้างอิงที่จำเป็นสำหรับกระแสข้อมูลที่ไม่ธรรมดาทุกกรณี

กฎเชิงปฏิบัติที่ยืนหยัดในการรีวิวโค้ดคือ ใช้ Signals สำหรับ state ภายในคอมโพเนนต์ที่อ่านในเทมเพลต และใช้ Observables สำหรับไปป์ไลน์อะซิงโครนัสที่ต้องใช้โอเปอเรเตอร์ ชั้น interop เชื่อมทั้งสองเข้าด้วยกันเพื่อไม่ให้โมเดลใดกลายเป็นกำแพงขวางกั้น

โอเปอเรเตอร์สำหรับการรวมยิ่งตอกย้ำประเด็นนี้ การประสานแหล่งข้อมูลอะซิงโครนัสหลายแหล่ง เช่น การผสานพารามิเตอร์ของ route เข้ากับฟิลเตอร์ที่ผู้ใช้บันทึกไว้และฟีดราคาแบบเรียลไทม์ คือสิ่งที่ combineLatest, withLatestFrom และ forkJoin ถูกสร้างมาเพื่อทำโดยแท้ Signals สามารถหาค่าที่คำนวณได้จาก signal อื่น แต่ไม่สามารถแสดงการประสานที่อิงเวลา เช่น "ปล่อยค่าออกมาเฉพาะหลังจากทั้งสามแหล่งได้ผลิตค่าอย่างน้อยหนึ่งค่าแล้ว" การประกอบเชิงประกาศตลอดช่วงเวลานั้นคือเหตุผลที่ยั่งยืนซึ่งทำให้ RxJS ยังคงอยู่ในกล่องเครื่องมือ

โอเปอเรเตอร์ RxJS หลักที่นักพัฒนา Angular ทุกคนพึ่งพา

โอเปอเรเตอร์คือฟังก์ชันบริสุทธิ์ที่รับ Observable เข้ามาแล้วคืน Observable ตัวใหม่ออกไป สถานการณ์สัมภาษณ์คลาสสิกคือ search-as-you-type เพราะมันฝึกโอเปอเรเตอร์สี่ตัวพร้อมกัน ได้แก่ การจำกัดอัตราอินพุต การข้ามค่าซ้ำ การยกเลิกคำขอที่ล้าสมัย และการกู้คืนจากข้อผิดพลาด

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
      )
    )
  );
}

รายละเอียดที่ละเอียดอ่อนซึ่งผู้สัมภาษณ์มักตรวจสอบคือ เหตุใด switchMap จึงอยู่ที่ระดับนอกสุดในขณะที่ catchError ซ้อนอยู่ภายใน การวาง catchError บน Observable ของ HTTP ที่อยู่ด้านในหมายความว่าคำขอที่ล้มเหลวจะคืนค่าเป็นอาร์เรย์ว่างและสตรีมด้านนอกยังคงไหลต่อไป การย้าย catchError ไปยังไปป์ด้านนอกจะยุติการค้นหาทั้งหมดหลังจากข้อผิดพลาดครั้งแรก ทำให้ทุกการกดแป้นครั้งต่อไปพังลง โอเปอเรเตอร์ flattening ทั้งสี่ตัวต่างตอบคำถามเรื่องการทำงานพร้อมกันที่ต่างกัน และการเลือกผิดตัวคือหนึ่งในสาเหตุที่พบบ่อยที่สุดของ race condition ในแอป Angular

การเลือกให้ถูกต้องนี้เองที่เปลี่ยนฟีเจอร์อันไม่แน่นอนให้กลายเป็นฟีเจอร์ที่คาดเดาได้ โมดูลโอเปอเรเตอร์ RxJS ของ SharpSkill ฝึกฝนความแตกต่างเหล่านี้ด้วยคำถามแบบให้คะแนนที่จำลอง race condition ที่โอเปอเรเตอร์แต่ละตัวป้องกันได้อย่างแม่นยำ

Subject และ multicasting: BehaviorSubject, ReplaySubject และ Subject

Observable ธรรมดาเป็นแบบ unicast กล่าวคือ แต่ละ subscriber ทำให้เกิดการทำงานใหม่ ส่วน Subject เป็นทั้ง Observable และ Observer ในตัว ซึ่งทำให้มันเป็นแบบ multicast คุณสมบัตินี้ทำให้ Subject เป็นวิธีมาตรฐานในการสร้าง state store แบบเบาหรือ event bus ภายใน service

ทั้งสามรูปแบบต่างกันที่สิ่งที่พวกมันเล่นซ้ำให้แก่ subscriber ที่มาสาย Subject ธรรมดาไม่ปล่อยอะไรจากอดีตออกมาเลย ดังนั้น subscriber จะเห็นเฉพาะค่าที่เข้ามาหลังจากที่มัน subscribe แล้ว BehaviorSubject เก็บค่าปัจจุบันไว้และเล่นซ้ำทันที นั่นคือเหตุผลที่มันต้องมีค่าเริ่มต้น ส่วน ReplaySubject บัฟเฟอร์จำนวนการปล่อยค่าในอดีตที่กำหนดค่าได้และเล่นซ้ำทั้งหมด

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
  }
}

มีนิสัยสองอย่างที่สำคัญตรงนี้ ประการแรก การเปิดเผย asObservable() แทนตัว Subject เองช่วยป้องกันไม่ให้โค้ดภายนอกเรียก next() และเปลี่ยนแปลง state ผ่านทางประตูหลัง ประการที่สอง การปล่อยการอ้างอิงอาร์เรย์ตัวใหม่แทนการกลายพันธุ์ในที่เดิมช่วยให้ change detection และการตรวจสอบความเท่ากันคาดเดาได้ รูปแบบนี้มีมาก่อน Signals และยังทำงานได้ดีสำหรับ state ที่ใช้ร่วมกันซึ่ง inject ได้และถูกบริโภคโดยหลายคอมโพเนนต์ที่ไม่เกี่ยวข้องกัน

อย่างไรก็ตาม BehaviorSubject ที่ใช้เพียงเพื่อเก็บค่าปัจจุบันเดี่ยว ๆ ที่อ่านในเทมเพลต ปัจจุบันมักแสดงออกได้ดีกว่าในรูปของ Signal แบบเขียนได้ ซึ่งช่วยขจัดการ subscribe และพิธีการ asObservable() ออกไป Subject ยังคงเป็นตัวเลือกที่แข็งแกร่งกว่าเมื่อ store จำเป็นต้องแปลง debounce หรือรวมการปล่อยค่าของมันกับสตรีมอื่นด้วย เพราะโอเปอเรเตอร์เหล่านั้นไม่มีสิ่งเทียบเท่าใน Signal โดยตรง การตัดสินใจเลือกระหว่างทั้งสองตามแต่ละกรณีการใช้งาน แทนที่จะใช้ตัวใดตัวหนึ่งเป็นค่าเริ่มต้นในทุกที่ คือเครื่องหมายของความเชี่ยวชาญที่ผู้สัมภาษณ์ให้คุณค่า

interop ของ RxJS กับ Signals ด้วย toSignal และ toObservable

แพ็กเกจ interop มาพร้อมสองฟังก์ชันหลัก toSignal แปลง Observable ให้เป็น Signal แบบอ่านอย่างเดียวที่เทมเพลตสามารถบริโภคได้โดยตรง และมันจัดการวงจรชีวิตของ subscription ให้โดยอัตโนมัติ toObservable ทำสิ่งตรงกันข้าม โดยเปลี่ยน Signal ให้เป็น Observable เพื่อให้โอเปอเรเตอร์ RxJS ประมวลผลการเปลี่ยนแปลงของมันได้

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 });
}

ตัวเลือก initialValue สมควรได้รับความสนใจ หากไม่มีมัน toSignal จะคืน Signal ที่มีชนิดเป็นค่าหรือ undefined เพราะยังไม่มีสิ่งใดถูกปล่อยออกมา การส่ง initialValue: null มอบ state เริ่มต้นที่ชัดเจนและชนิดที่สะอาดกว่าให้เทมเพลต สำหรับแหล่งข้อมูลแบบซิงโครนัสอย่าง BehaviorSubject นั้น requireSync: true รับประกันว่ามีค่าพร้อมใช้ตั้งแต่การอ่านครั้งแรกและตัด undefined ออกจากชนิดโดยสิ้นเชิง สะพานสองทิศทางนี้เองที่ทำให้โค้ดเบสสามารถนำ Signals มาใช้อย่างค่อยเป็นค่อยไปโดยไม่ต้องเขียนไปป์ไลน์ Observable เดิมทุกตัวใหม่ โมดูลสัมภาษณ์ Angular Signals ที่มาคู่กันครอบคลุมโมเดล reactivity อย่างละเอียด และ ภาพรวมฟีเจอร์ Angular 18 Signals ติดตามว่า primitive เหล่านี้วิวัฒน์มาอย่างไร

เชื่อม Observables สู่ signal resource ด้วย rxResource

Angular 20 ผลักดัน rxResource ให้เป็นวิธีเชิงประกาศในการโหลดข้อมูลอะซิงโครนัสจาก Observable เข้าสู่ state ที่อิง signal มันติดตามสถานะ loading, error และ resolved เป็น signal ซึ่งช่วยขจัดงานจัดการ subscription ด้วยมือส่วนใหญ่ที่การเรียก HTTP เคยต้องทำ

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
}

การเปลี่ยนชื่อจาก request/loader ใน Angular 19 เป็น params/stream ใน Angular 20 เป็นจุดที่มักพลาดกันบ่อย ดังนั้นการเอ่ยถึงชื่อฟิลด์ปัจจุบันจึงบ่งบอกถึงความรู้ที่ทันสมัยในการสัมภาษณ์ เมื่อ signal category เปลี่ยนแปลง rxResource จะยกเลิกคำขอก่อนหน้าและเริ่มสตรีมใหม่ มอบความหมายแบบ switchMap ให้ฟรี ๆ เอกสารอ้างอิง API ของ rxResource บันทึกวงจรชีวิตของสถานะไว้อย่างครบถ้วน

หลีกเลี่ยงการรั่วไหลของหน่วยความจำด้วย takeUntilDestroyed

subscribe() ด้วยมือใด ๆ ที่มีอายุยืนกว่าคอมโพเนนต์ของมันจะทำให้หน่วยความจำรั่วไหลและอาจยิง callback ไปยัง view ที่ถูกทำลายไปแล้ว วิธีแก้ในอดีตคือรูปแบบ takeUntil(this.destroy$) พร้อม Subject ที่ถูกทำให้เสร็จสมบูรณ์ใน ngOnDestroy Angular แทนที่ boilerplate นั้นด้วย takeUntilDestroyed ซึ่งเป็นโอเปอเรเตอร์ที่ผูก subscription เข้ากับ injection context ปัจจุบันหรือ DestroyRef ที่ระบุอย่างชัดเจน

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));
  }
}

เมื่อเรียกภายใน constructor หรือการกำหนดค่าเริ่มต้นของฟิลด์ takeUntilDestroyed() ไม่ต้องการอาร์กิวเมนต์เพราะมันอ่าน DestroyRef ที่อยู่รอบ ๆ เมื่อเรียกที่อื่น เช่น ภายในเมธอด มันต้องการ DestroyRef ที่ระบุอย่างชัดเจนส่งเข้ามา ดังที่แสดงข้างต้น การให้ความสำคัญกับ toSignal และ rxResource ช่วยหลีกเลี่ยง subscription ด้วยมือโดยสิ้นเชิง ซึ่งเป็นวิธีที่สะอาดที่สุดในการเลี่ยงการรั่วไหล เมื่อจำเป็นต้องใช้ subscription แบบดิบจริง ๆ takeUntilDestroyed คือเครื่องป้องกันที่ถูกต้อง

คำถามสัมภาษณ์ RxJS Angular ที่ควรค่าแก่การซ้อม

ผู้สัมภาษณ์มักวนเวียนอยู่กับหัวข้อที่ให้สัญญาณสูงชุดเดิม ๆ การตอบคำถามเหล่านี้ได้อย่างกระชับช่วยแยกผู้สมัครที่เคยใช้ RxJS ออกจากผู้ที่เพียงอ่านเกี่ยวกับมัน

• Hot กับ cold Observables cold Observable เริ่ม producer ของมันต่อการ subscribe แต่ละครั้ง ดังนั้น subscriber แต่ละรายจึงได้การทำงานที่เป็นอิสระ (คำขอ HTTP ใหม่) ส่วน hot Observable ใช้ producer เดียวร่วมกันในทุก subscriber ซึ่งเป็นสิ่งที่ Subject และ share() สร้างขึ้น
• switchMap กับ mergeMap กับ concatMap กับ exhaustMap switchMap ยกเลิก Observable ด้านในตัวก่อนหน้า (เหมาะสำหรับการค้นหา) mergeMap รัน Observable ด้านในทั้งหมดพร้อมกัน concatMap เข้าคิวตามลำดับ exhaustMap เพิกเฉยต่ออินพุตใหม่ขณะที่ตัวหนึ่งกำลังรันอยู่ (เหมาะสำหรับป้องกันการ submit ฟอร์มซ้ำสองครั้ง)
• วิธีป้องกันการรั่วไหลของหน่วยความจำ ให้ความสำคัญกับ toSignal, rxResource หรือ pipe async ซึ่งทั้งหมดยกเลิกการ subscribe โดยอัตโนมัติ สำหรับ subscription ด้วยมือ ให้ใช้ takeUntilDestroyed
• เมื่อใดควรแปลง Observable เป็น Signal เลือกใช้ toSignal เมื่อเทมเพลตต้องการค่าแบบซิงโครนัสในขณะที่แหล่งข้อมูลเป็นอะซิงโครนัส ทำให้ change detection อ่านค่าล่าสุดได้โดยไม่ต้องใช้ pipe async

การทำแบบฝึกหัดเวอร์ชันให้คะแนนของคำถามเหล่านี้ภายใต้ความกดดันด้านเวลาคือวิธีที่เร็วที่สุดในการซึมซับมัน ศูนย์รวมเทคโนโลยี Angular จัดกลุ่มโมดูล RxJS, Signals และ change detection ที่แมปตรงกับคำถามเหล่านี้

บทสรุป

RxJS และ Signals เป็นหุ้นส่วนกันในสแตก Angular สมัยใหม่ ไม่ใช่คู่แข่ง ประเด็นสำคัญที่ควรค่าแก่การนำติดตัวไปยังโปรเจกต์หรือการสัมภาษณ์ครั้งต่อไป

• ใช้ Signals สำหรับ state แบบซิงโครนัสของคอมโพเนนต์ และ Observables สำหรับไปป์ไลน์เหตุการณ์อะซิงโครนัสที่ต้องใช้โอเปอเรเตอร์
• เลือก switchMap เพื่อยกเลิกงานที่ล้าสมัย exhaustMap เพื่อบล็อกการทำซ้ำ และคง catchError ไว้บน Observable ด้านในเพื่อให้สตรีมด้านนอกอยู่รอดจากความล้มเหลว
• เปิดเผย state ของ Subject ผ่าน asObservable() และปล่อยค่าที่ไม่เปลี่ยนแปลงได้เพื่อให้ change detection คาดเดาได้
• เชื่อมสองโมเดลด้วย toSignal และ toObservable และส่ง initialValue หรือ requireSync เพื่อให้ได้ชนิดที่สะอาด
• โหลดข้อมูลอะซิงโครนัสเชิงประกาศด้วย rxResource ใน Angular 20 โดยจดจำชื่อฟิลด์ params และ stream
• ป้องกันทุก subscription ด้วยมือที่ยังเหลืออยู่ด้วย takeUntilDestroyed เพื่อขจัดการรั่วไหล