如何只订阅一次可观察的

Question

我正在尝试订阅两个 observable 并获取要存储在数组对象中的值。它工作正常，但我的问题是它重复了三次，我不明白。我在一项服务中这样做是为了创建一项新服务。下面是代码示例。我也想知道我可以使用 Promise 而不是 Angular 中的 observable 吗？或者我可以在获得价值后将 observable 转换为 promise 并解决？谢谢你的帮助

 addEmployeeData() {
    const employeeObservable = this.apiService.getEmployeeDataObservable();
    employeeObservable.subscribe({
      next: (res: any) => {
        let employee = res;
        const paramsObservable = this.apiService.getParamsObservable();
        pageParamsObservable.subscribe({
          next: (pageParams: any) => {

Answer 1

是的，您可以像使用 Promises 一样使用 Observables：

async asyncAddEmployeeData(): Promise<any> {
  return this.apiService.getEmployeeDataObservable()
    .pipe(
      mergeMap(employeeData => this.apiService.getParamsObservable()
        .pipe(
          tap((paramsData): void => {
            // There is available data
            // from apiService.getEmployeeDataObservable()
            // as employeeData variable
            // and data from apiService.getParamsObservable()
            // as paramsData.
            // You can do in tap function all the same
            // as in next in the subscribe.
          }),
        )
      ),
    )
    .toPromise();
}

并像这里一样使用它：

async ngOnInit(): Promise<void> {
  // ngOnInit just for example.
  const someVariable = await this.asyncAddEmployeeData();
}

但是使用 Observable 的常规方式是这样的：

addEmployeeData(): Observable<any> {
  return this.apiService.getEmployeeDataObservable()
    .pipe(
      mergeMap(employeeData => this.apiService.getParamsObservable()
        .pipe(
          tap(paramsData => {
            // There is available data
            // from apiService.getEmployeeDataObservable()
            // as employeeData variable
            // and data from apiService.getParamsObservable()
            // as paramsData.
          }),
        )
      ),
      take(1), // Just if you need only first value, if not, please, remove this string.
    );
}

和订阅：

ngOnInit(): void {
  // ngOnInit just for example.
  this.subscription = this.addEmployeeData().subscribe();
}

不要忘记取消订阅以避免内存泄漏：

ngOnDestroy(): void {
  this.subscription.unsubscribe();
}

Answer 2

由于您有 2 个 Observable，并且您仅在订阅第二个 observable 后才执行实际逻辑，我假设没有第二个 observable 数据，您没有实现任何逻辑。

为此，您有 2 个选择。

1. 使用 mergeMap RxJS 运算符并按照 Mikhail Filchushkin 上面所说的操作。

（或）

1. 我的方法，使用combineLatest 运算符并在订阅完成后使用Subject 变量销毁订阅。你就是这样做的：

与mergeMap 相比，此方法的优势在于您将只有一个订阅和一个取消订阅，而在mergeMap 中您将拥有两个。

MergeMap 将订阅第一个变量，如果它成功订阅，那么只有它会订阅第二个订阅。

在combineLatest 中，无论即将到来的数据如何，它都会订阅。

这取决于你如何使用它，这是一种优势和劣势。

如果您只想要第一个值，请使用take(1) 运算符阻止它进一步订阅

const employeeObservable = this.apiService.getEmployeeDataObservable();
const paramsObservable = this.apiService.getParamsObservable();

destroy$: Subject<any> = new Subject(); // This is used to destroy the subscription later

// Combine the observables and destroy once done in 1 shot.

combineLatest(employeeObservable,paramsObservable)
  .pipe(
      takeUntil(this.destroy$),
      take(1)
   )
  .subscribe(
     ([employeeObservableData, paramsObservableData]) => {
        if (employeeObservableData && paramsObservableData) {
          // do your logic here
        }  
     }
  );

ngOnDestroy() {
   this.destroy$.next();
   this.destroy$.complete();
}

Answer 3

1. 您可以使用任何 RxJS 高阶映射运算符（如 switchMap）从一个 observable 切换到另一个相互依赖的 observable。

addEmployeeData() {
  this.apiService.getEmployeeDataObservable().pipe(
    switchMap(_ => {
      let employee = res;  // why is this needed though?
      return this.apiService.getParamsObservable();
    })
  ).subscribe({
    next: (pageParams: any) => { 
      // handle response
    },
    error: (error: any) => {
      // handle error
    }
  );
}

1. 如果 observables 彼此不相关（例如，如果第二个请求不依赖于第一个请求的结果），您可以使用 RxJS forkJoin 或 combineLatest 等函数来并行触发 observables。

   有关映射运算符和组合函数的快速比较，请参阅我的帖子 here。

2. 最好关闭ngOnDestroy 中的任何开放订阅，这样它们就不会持续存在并导致潜在的内存泄漏。有多种方法可以关闭订阅。

   3.1 unsubscribe - 将订阅分配给成员变量并在 ngOnDestroy 挂钩中对其调用 unsubscribe()。

   3.2 take(1) - 在发出第一个通知后关闭订阅。

   3.3 first() - 与take(1) 类似，但如果没有任何发射与谓词匹配，则需要额外的谓词并发出错误。见here for take(1) vs first()。

   3.4。 takeUntil() - 使用额外的多播以单次发射关闭多个打开的订阅。最喜欢的方式是它的优雅。请参阅here 了解更多信息。

注意：Angular HttpClient 返回的 Observables 在第一次发射后完成，无需任何上述操作。所以在大多数情况下它们不是必需的。

Answer 4

尝试在 onDestroy 上创建订阅并取消订阅。

// import 
import { Subscription } from 'rxjs';

subscription: Subscription;

ngOnInit() { 
    this.subscription = // employeeObservable.subscribe()
}

ngOnDestroy() { 
    this.subscription.unsubscribe();
}