时间戳

首页 > 解决方案 > 如何在 rxjs 中停止可观察的共享

angular - 如何在 rxjs 中停止可观察的共享

问题描述

我想停止 observable，它是由 share 运算符创建的。

我的服务：

private timer = interval(1000).pipe(take(60), share());
private timerStarted = false;

startTimer() {
  this.timer.pipe(
    tap(() => this.timerStarted = true),
    finalize(() => this.timerStarted = false)
  ).subscribe();
}

getTimer() {
  return this.timer;
}

getTimerStarted() {
  return this.timerStarted;
}

我的组件：

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

ngOnInit() {
  if (this.timerService.getTimerStarted()) {
    this.timerService.getTimer().pipe(
      takeUntil(this.destroy$),
    ).subscribe();
  }
}

onStartTimer() {
  this.timerService.startTimer();
  this.timerService.getTimer().pipe(
    takeUntil(this.destroy$),

  ).subscribe();
}

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

我尝试添加到服务停止方法：

stopTimer() {
        this.timer.unsubscribe();
    }

并在 ngOnDestroy 中使用它：

ngOnDestroy() {
  this.destroy$.next(true);
  this.timerService.stopTimer();
}

我想停止在后台创造价值。

标签： angularrxjsobservable

解决方案

您不需要在您的服务中创建订阅，目前尚不清楚您为什么要这样做。share() 一旦没有订阅者，它自然会终止，但在服务中，您要确保一个订阅者始终存在。只是这样做：

export class TimerService {
  timer$ = interval(1000).pipe(
             take(60), 
             tap(() => this.timerStarted = true), 
             finalize(() => this.timerStarted = false),
             share());
  private timerStarted = false;

  getTimerStarted() {
    return this.timerStarted;
  }
}

@Component({
  selector: 'my-app',
  templateUrl: './app.component.html',
  styleUrls: [ './app.component.css' ],
  providers: [TimerService]
})
export class AppComponent  {
  name = 'Angular';

  sub;
  constructor(private timer: TimerService) {
    this.sub = this.timer.timer$.subscribe(v => console.log(v));
    setTimeout(() => this.sub.unsubscribe(), 4000)
  }
}

在日志中，您会看到计时器发出 4 次，然后在超时取消订阅后停止。您可以轻松地取消订阅 onDestroy 挂钩或使用 destroy$ 发射方法。两者都会产生相同的效果。

闪电战：https ://stackblitz.com/edit/angular-9a9bkl?file=src/app/app.component.ts

根据评论编辑：

由于您确实希望在组件之间保持订阅有效，因此您需要修改您的服务以允许您结束该订阅：

private timer = interval(1000).pipe(take(60), takeUntil(stopTimer$), share());
private timerStarted = false;
private stopTimer$ = new Subject();

startTimer() {
  this.timer.pipe(
    tap(() => this.timerStarted = true),
    finalize(() => this.timerStarted = false)
  ).subscribe();
}

stopTimer() {
  this.stopTimer$.next();
}

此方法将结束服务级别订阅和所有其他订阅，因为 takeUntil 是计时器管道本身的一部分。

推荐阅读