如何优化Angular Observable链式调用？解决组件重复API请求问题

Hey there! Let's break down what's causing those unwanted extra API requests and refactor your code to be cleaner, more maintainable, and leak-free.

What's Going Wrong? #

Your current setup has a few key issues that lead to extra requests and potential memory leaks:

• Nested subscriptions: Every time the timer fires, you create a new subscription to getPreferencesAsObservable(). If this observable emits frequently or isn't properly cleaned up, you end up with overlapping subscriptions triggering duplicate API calls.
• Unmanaged subscriptions: The inner subscriptions for your order requests (and pagination calls) aren't tracked or canceled. This means they can keep running even after the component is destroyed, leading to extra requests and memory leaks.
• Manual pagination logic: Your separate getNext... methods add more nested subscriptions and make it hard to control the flow of pagination requests.

Optimized Solution with RxJS Best Practices #

Let's rewrite this using RxJS operators to eliminate nested subscriptions, manage all subscriptions properly, and handle pagination more elegantly.

First, add a destroy subject to clean up subscriptions when the component is removed:

import { Subject, EMPTY, forkJoin } from 'rxjs';
import { timer, switchMap, filter, mergeMap, expand, map, reduce, takeUntil } from 'rxjs/operators';

// ...

private destroy$ = new Subject<void>();

ngOnInit(): void { 
  timer(0, 120 * 1000)
    .pipe(
      // Switch to preferences observable, canceling any previous pending preference subscriptions
      switchMap(() => this.shopService.getPreferencesAsObservable()),
      // Only proceed if preferences exist
      filter(preferences => !!preferences),
      // Parallelize both initial order requests
      mergeMap(() => forkJoin([
        this.fetchAllPendingSlotsOrders(),
        this.fetchAllPendingNoSlotsOrders()
      ])),
      // Clean up all subscriptions when component is destroyed
      takeUntil(this.destroy$)
    )
    .subscribe({
      next: ([allSlotsOrders, allNoSlotsOrders]) => {
        // Process all your order data here
        this.processSlotsOrders(allSlotsOrders);
        this.processNoSlotsOrders(allNoSlotsOrders);
      },
      error: (err) => {
        // Handle errors (e.g., log to service, show user message)
        console.error('Failed to fetch orders:', err);
      }
    });
}

// Fetch ALL pages of slot orders in one observable stream
private fetchAllPendingSlotsOrders() {
  return this.apiService.fetchShopOrders("status=PENDING&only_slots=true")
    .pipe(
      // Automatically fetch next pages until there are none left
      expand(orders => {
        if (orders.next_page) {
          return this.apiService.fetchNextSetOfOrders(orders.next_page);
        }
        return EMPTY; // End the stream when no more pages
      }),
      map(orders => orders.data), // Extract the order data from each response
      // Combine all pages into a single array (remove if you want to process page-by-page)
      reduce((accumulatedOrders, currentPageOrders) => [...accumulatedOrders, ...currentPageOrders], [])
    );
}

// Fetch ALL pages of non-slot orders
private fetchAllPendingNoSlotsOrders() {
  return this.apiService.fetchShopOrders("status=PENDING")
    .pipe(
      expand(orders => {
        if (orders.next_page) {
          return this.apiService.fetchNextSetOfOrders(orders.next_page);
        }
        return EMPTY;
      }),
      map(orders => orders.data),
      reduce((accumulatedOrders, currentPageOrders) => [...accumulatedOrders, ...currentPageOrders], [])
    );
}

// Process your slot order data here
private processSlotsOrders(orders: any[]) {
  // Your existing order processing logic goes here
}

// Process your non-slot order data here
private processNoSlotsOrders(orders: any[]) {
  // Your existing order processing logic goes here
}

ngOnDestroy(): void {
  // Trigger cleanup of all subscriptions
  this.destroy$.next();
  this.destroy$.complete();
}

Key Improvements Explained #

• switchMap: Ensures that if a new timer event fires before the previous preference/order requests complete, the old requests are canceled. This prevents overlapping requests and extra calls.
• filter: Skips any timer cycles where preferences aren't configured, so you never waste API calls on invalid states.
• forkJoin + mergeMap: Runs both initial order requests in parallel, which is more efficient than sequential calls.
• expand: Handles pagination automatically without manual getNext... calls. It keeps fetching pages until there are no more, then ends the stream.
• takeUntil: Guarantees all subscriptions are canceled when the component is destroyed, eliminating memory leaks and stray API requests.
• No nested subscriptions: All logic is chained through RxJS operators, making the code easier to read, debug, and maintain.

Optional Adjustments #

• If you want to process each page of orders as it comes in (instead of waiting for all pages), remove the reduce operator and handle each page in the next callback.
• If you need to wait for the previous batch of requests to finish before starting the next timer cycle, replace switchMap with concatMap.

内容的提问来源于stack exchange，提问作者Manuel Brás