Angular组件中RxJS订阅使用及数组Observable合并问题咨询

• 订阅泄漏问题：当前ngOnDestroy中的取消订阅逻辑被注释，且每次调用search()、showMore()方法时，都会直接覆盖searchSub、showMoreSub变量，之前未取消的订阅会一直保留在内存中，直到应用关闭，属于典型的内存泄漏。
• distinctUntilChanged无效：你将该操作符放在单次搜索请求的Observable管道中，每次搜索都会生成全新的Observable，该操作符无法对比不同搜索请求的返回值，完全没有起到去重作用。
• 订阅管理逻辑不规范：没有要求订阅必须在constructor中声明，但是需要保证所有订阅都能被追踪到，当前直接覆盖订阅变量的写法会导致旧订阅丢失追踪，无法取消。

基础修复方案（改动最小） #

你可以用Subscription的add方法统一管理所有订阅，不需要单独声明多个订阅变量：

import {Component, OnDestroy} from '@angular/core';
import {ISearchParams} from '../../models/search-params.interface';
import {ShowService} from '../../services/show.service';
import {IResultsIds} from '../../models/results.interface';
import {Subscription} from 'rxjs';

@Component({
  selector: 'app-search',
  templateUrl: './search-tab.component.html',
  styleUrls: ['./search-tab.component.scss']
})
export class SearchTabComponent implements OnDestroy {
  searchParams!: ISearchParams;
  searchResults!: IResultsIds;
  // 统一管理所有订阅
  private subs = new Subscription();

  constructor(private movieService: ShowService) {}

  search(searchParams: ISearchParams): void {
    this.searchParams = searchParams;
    // 新订阅添加到subs中统一管理
    const searchSub = this.movieService.search(searchParams)
      .subscribe(results => {
        this.searchResults = results;
      });
    this.subs.add(searchSub);
  }

  showMore(): void {
    if (!this.isFinished()) {
      this.searchParams.page++;
      const showMoreSub = this.movieService.search(this.searchParams)
        .subscribe(results => {
          this.searchResults.ids.push(...results.ids);
        });
      this.subs.add(showMoreSub);
    }
  }

  isFinished = () => this.searchParams.page >= this.searchResults.total_pages;

  ngOnDestroy(): void {
    // 一次性取消所有订阅
    this.subs.unsubscribe();
  }
}

如果想进一步用更符合RxJS设计理念的写法，可以用async管道，不需要手动管理订阅，Angular会在组件销毁时自动取消订阅：

// 声明结果数组的Observable
searchResults$!: Observable<IResultsIds>;
// 用Subject触发加载更多动作
private loadMore$ = new Subject<void>();

search(searchParams: ISearchParams): void {
  this.searchParams = searchParams;
  // 每次新搜索重置流，用scan累计分页结果
  this.searchResults$ = this.movieService.search(searchParams).pipe(
    switchMap(firstPage => 
      this.loadMore$.pipe(
        switchMap(() => {
          this.searchParams.page++;
          return this.movieService.search(this.searchParams);
        }),
        // 累计所有页的id
        scan((acc, curr) => {
          acc.ids.push(...curr.ids);
          return acc;
        }, firstPage),
        startWith(firstPage)
      )
    )
  )
}

showMore(): void {
  if (!this.isFinished()) {
    this.loadMore$.next();
  }
}

对应的HTML也可以直接用async管道：

<main class="main-content search-container">
  <app-search-form (searchParams)="search($event)"></app-search-form>
  <div class="results" *ngIf="searchResults$ | async as searchResults">
    <app-show-description *ngFor="let id of searchResults.ids"
                          [showType]="searchParams.type"
                          [showId]="id"></app-show-description>
  </div>
  <button *ngIf="(searchResults$ | async) && !isFinished()"
          (click)="showMore()"
          class="load-btn more-btn">show more...</button>
</main>

你用concat的写法错误是因为concat只是按顺序拼接多个Observable的发射事件，不会对事件的内容做处理，所以你会收到两次独立的数组发射，而不是合并后的单个数组。
你需要的是累计多次Observable的发射值，把数组合并成一个，应该用scan操作符，它和数组的reduce类似，会保留之前的累计值，每次新发射值进来的时候可以自定义合并逻辑。

正确写法 #

// 先声明触发加载更多的Subject
private loadMoreAction$ = new Subject<ISearchParams>();
// 结果流，累计所有页的id
distinctMovies$ = this.loadMoreAction$.pipe(
  switchMap(params => this.movieService.search(params).pipe(pluck('ids'))),
  // 累计所有返回的id数组
  scan((allIds, currIds) => [...allIds, ...currIds], [])
);

// 首次搜索调用
search(searchParams: ISearchParams): void {
  this.searchParams = searchParams;
  // 搜索的时候重置页码，给Subject传参触发请求
  this.searchParams.page = 1;
  this.loadMoreAction$.next(this.searchParams);
}

// 加载更多调用
showMore(): void {
  if (!this.isFinished()) {
    this.searchParams.page++;
    this.loadMoreAction$.next(this.searchParams);
  }
}

这时候distinctMovies$发射的就是所有页合并后的id数组，你可以直接用async管道在模板中遍历。

内容的提问来源于stack exchange，提问作者Yurii Hrecheniuk