在 Angular 2 的 rxjs 中使用 switchMap 运算符时，我们是否必须取消订阅？

Question

在 Angular 2 中有一些你不需要取消订阅的 observables。例如http请求和activatedRoute.params。

Angular/RxJs When should I unsubscribe from `Subscription`

但是，当我使用 switchMap 时会发生什么，例如在activatedRoute.params 中，我在该 switchMap 中访问了一个服务，该服务返回一个 observable，如果以通常的方式订阅则需要取消订阅。

类似这样的：

this.activatedRoute.params
    .switchMap((params: Params) => this.userService.getUser(+params['id']))
    .subscribe((user: User) => this.user = user);

如果我在没有 switchMap 和 activateRoute.params 的情况下调用 this.userService，我将不得不取消订阅。

// userService.getUser() takes in optional id?: number.
this.subscription = this.userService.getUser().subscribe(
    (user: User) => {
        this.user = user;
    }
 );

后来……

this.subscription.unsubscribe();

我的问题是，如果我在其上使用 switchMap 并调用需要取消订阅的服务，我是否需要取消订阅 activateRoute.params ？

Answer 1

新的 RxJS 文档解释了 switchMap 何时会继续侦听以及何时会在异常传播期间停止。

参见 iWillContinueListening 和 iWillStopListening 演示：

https://www.learnrxjs.io/learn-rxjs/operators/error_handling/catch

Answer 2

switchMap 在前一个和新的 observable 之间创建一个链接。如果你改变了第一个 observable，第二个总是会被触发。

在switchMap 之后订阅的任何内容都将听到初始 observable 和返回的 observable 的变化。

使用take、takeUntil 或takeWhile 完全停止第一个可观察对象以更新第二个观察对象或其余部分。喜欢：

const howTimerWorks = interval(5000).pipe(
  take(2), // only get 2 responses after 5 seconds each
  switchMap(initialNumber => interval(1000)));

// 0 after 5s, then 1, 2 , 3, (new Subs) 0, 1, ... every sec, forever now.
howTimerWorks.subscribe(console.log)

Answer 3

一旦你做了一个 switchmap，Subscription 就会附加到最后一个 Observable。如果第一个 observable 继续触发并不重要。 switchmap 块只执行一次。

如果它永远不会关闭，您必须从最后一个取消订阅。

检查此代码：

import { Component, OnInit } from '@angular/core';
import { Observable, BehaviorSubject, Subscription} from 'rxjs';
import * as Rx from 'rxjs';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit {
  title = 'app works!';
  private source: Rx.Subject<any>;
  private source2: Rx.Subject<any>;
  private composed: Rx.Observable<any>;
  private composedSub: Subscription;

  public ngOnInit(): void {
    this.source = new Rx.Subject();
    this.source2 = new Rx.Subject();

    this.composed = this.source.switchMap(value => this.source2);

    this.composedSub = this.composed.subscribe(value => console.log(value));
    console.log(this.composedSub);
  }

  private onClick() {
    // Triggers the first observable, the console.log is never executed.
    this.source.next(1);
  }

  private onClick2() {
    // Console.log is executed, prints "1"
    this.source2.next(1);
  }

  private onClick3() {
    // The console log is never called again after click on this button.
    this.composedSub.unsubscribe();
  }

  private onClick4() {
    // The first observable finish. the console log keeps printing unles onClick3 is executed
    this.source.complete();
  }

  private onClick5() {
    // console.log never executed egain.
    this.source2.complete();
  }
}

Answer 4

如果您订阅的 observable 源始终完成或出错，则无需取消订阅。

但是，如果您使用 switchMap 从源中组合另一个 observable，则是否需要取消订阅取决于 switchMap 中返回的 observable。如果返回的 observable 并不总是完成或出错，那么，是的，你需要取消订阅。

如果来源出错，会自动退订：

const source = new Rx.Subject();
const composed = source.switchMap(() => Rx.Observable.interval(200));

composed.subscribe(value => console.log(value));
source.next(1);

setTimeout(() => {
  console.log("Erroring...");
  source.error(new Error("Boom!"));
}, 1000);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5.4.1/bundles/Rx.min.js"></script>

但是，如果源完成，则不会发生自动退订：

const source = new Rx.Subject();
const composed = source.switchMap(() => Rx.Observable.interval(200));

composed.subscribe(value => console.log(value));
source.next(1);

setTimeout(() => {
  console.log("Completing...");
  source.complete();
}, 1000);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5.4.1/bundles/Rx.min.js"></script>