如何取消订阅嵌套的 observable答案

【问题标题】：How to unsubscribe from nested observable如何取消订阅嵌套的 observable
【发布时间】：2020-03-29 00:39:12
【问题描述】：

在我们的 Angular 组件上使用 ngOnDestroy() 方法，如果在离开页面时它们仍然处于未决状态，我们会取消 http 请求。在某些页面上，我们使用自定义的通用缓存帮助器来防止重新加载已经加载的数据。

import { HttpClient } from "@angular/common/http";
import { Injectable } from "@angular/core";
import { AsyncSubject } from "rxjs";

@Injectable()
export class HttpCacheHelper {
    public cache = new Map<string, AsyncSubject<any>>();

    public constructor(private readonly http: HttpClient) {        

    }

    public get<T>(url: string): AsyncSubject<T> {
        if (!this.cache.has(url)) {
            const subject = new AsyncSubject<T>();
            this.cache.set(url, subject);
            this.http.get(url)
                .subscribe((data:any) => {
                    subject.next(data as T);
                    subject.complete();
                });
        }
        return this.cache.get(url);
    }
}

如果我取消订阅 AsyncSubject，我的 http 调用（当然）不会取消。如何做到这一点？

【问题讨论】：

你不能只使用共享操作符：http.get().pipe(share())？
您应该考虑选择一个接受的答案，以帮助可能看到此帖子的其他人

标签： angular typescript rxjs observable

【解决方案1】：

在这种情况下，您不必取消订阅 http 源（请参阅 this question)。通常您可以使用 take 运算符或平面图进行嵌套订阅。

【讨论】：

据我了解，它会在请求完成后取消订阅。就我而言，我想在离开组件时退订。
为什么不把 this.http.get(url) 放到缓存中？
我第二次在 http.get 订阅时重复调用。
@Marcel-Hoekstra 提到你可以使用共享运算符
共享运算符是否可以处理已完成的呼叫？在这种情况下，我将它用于缓存，并且可以在进行原始调用后的几分钟内使用缓存。

【解决方案2】：

获取订阅对象并取消订阅。

import { HttpClient } from "@angular/common/http";
import { Injectable } from "@angular/core";
import { AsyncSubject, Subscription } from "rxjs";

@Injectable()
export class HttpCacheHelper {
    public cache = new Map<string, AsyncSubject<any>>();
    private mySubscription: Subscription;
    public constructor(private readonly http: HttpClient) {        

    }

    public get<T>(url: string): AsyncSubject<T> {
        if (!this.cache.has(url)) {
            const subject = new AsyncSubject<T>();
            this.cache.set(url, subject);
    this.mySubscription = this.http.get(url)
                .subscribe((data:any) => {
                    subject.next(data as T);
                    subject.complete();
                });
        }
        return this.cache.get(url);
    }

// Use below function when required
ngOnDestroy(): void {
    this.mySubscription.unsubscribe();
  }

}

【讨论】：

【解决方案3】：

您需要取消订阅源：

public get<T>(url: string): AsyncSubject<T> {
    if (!this.cache.has(url)) {
        const subject = new AsyncSubject<T>();

        const subscription = this.http.get(url)
            .subscribe((data:any) => {
                subject.next(data as T);
                subject.complete();
            });

         this.cache.set(url, {
           subscribe(observer) { subject.subscribe(observer) },
           unsubscribe(observer) { subscription.unsubscribe() }
         });
    }

    return this.cache.get(url);
}

【讨论】：

这看起来很有希望！我可以使用哪种类型的缓存映射？新地图();
我觉得可以Observable

【解决方案4】：

您也可以尝试以这种方式取消订阅

创建一个我们想用来取消订阅的主题

private onDestroy$ = new Subject();

在您需要取消订阅的任何 .subscribe() 之前添加它

.pipe(takeUntil(this.onDestroy$))

例子

this.anyService.getData()
  .pipe(takeUntil(this.onDestroy$))
  .subscribe((data: any) => {
    // Do work ...
  });

然后像这样使用ngOnDestroy()

ngOnDestroy() {
  this.onDestroy$.next();
  this.onDestroy$.complete(); // Also unsubscribe from the subject itself
}

当我们 ngOnDestroy() 运行并完成主题时，任何使用 takeUntill 的订阅也将自动取消订阅

【讨论】：

【解决方案5】：

我会选择publishReplay 和refCount：

@Injectable()
export class HttpCacheHelper {
  public cache = new Map<string, Observable<yourType>>();

  public constructor(private readonly http: HttpClient) {}

  public get<T>(url: string): Observable<T> {
    if (!this.cache.has(url)) {
      const obs = this.http.get(url).pipe(
         publishReplay(1),
         refCount()
      )

      this.cache.set(url, obs); 
    }

    return this.cache.get(url);
  }

  // Use below function when required
  ngOnDestroy(): void {
    this.mySubscription.unsubscribe();
  }
}

【讨论】：