可观察订阅不执行答案

【问题标题】：Observable Subscribe Doesnt Execute可观察订阅不执行
【发布时间】：2021-06-08 16:07:03
【问题描述】：

我正在尝试一次将列表中的消息显示 1 秒钟。我无法使用interval(1000)，因为我需要能够在满足条件后立即清除消息并显示下一条消息 1 秒。

目前为止

nextMessage() : Observable<any>{
    this.currentMessage.next(this.messages[this.messagePointer++]);
    return this.currentMessages;
}

subscribeMessage() : void{
    this.currentMessage.pipe(switchMap(_=timer(1000).pipe(map(x=>this.nextMessage())))).subscribe(x=>console.log(x));
}

目前这不会产生任何结果。我的想法是，nextMessage 例程每 1 秒更新一次 currentMessage Observable，如果它在两者之间被外部源更新，则 Timer 将被取消并重新启动。 subscribe 方法目前不输出任何内容，但是如果我尝试这样做

this.currentMessage.pipe(tap(x=>console.log(x)),switchMap(_=timer(1000).pipe(map(x=>this.nextMessage())))).subscribe(x=>console.log(x));

我每隔 1 秒从我的数据输出中看到值。为什么点击有效，但订阅无效。

编辑

这是StackBlitz Demo的链接

【问题讨论】：

抱歉，currentMessages 是什么？无论如何，如果您可以提供完整的课程以便可以重现这将是很棒的，因为我真的认为这还不足以重现这个问题。还有什么_= 我认为这是一个错字？最好的办法是，如果您可以与那里的完整示例共享一个 stackblitz，您可以这样做吗？
另外tap 可能有效，因为它位于pipe 的开头，所以它只是获取通过currentMessage 流的新消息，而在tap 你之后有那个switchMap 这可能会破坏事情。
@DarioPiotrowicz 我添加了一个 StackBlitz 演示，展示了我的体验。我无法上传我的整个项目，但这就是与此问题相关的全部内容
太棒了，谢谢:)，在你添加了stackblitz之后，我可以看到有人已经给出了一个全面的答案，所以我想我不再需要了:'(

标签： typescript rxjs observable

【解决方案1】：

问题：switchMap 在内部可观察对象发射之前杀死可观察对象。

流程如下：

currentMessage 发出消息
tap 记录该消息
switchMap 启动计时器(1000)
当定时器触发时，你调用nextMessage()
nextMessage 然后将一个值推送到currentMessage
在移至 nextMessage 的下一行之前，currentMesage 会发出一条消息（与第 1 步相同）
tap 记录该消息（与第 2 步相同）
switchMap 杀死之前的 observable，并启动一个新的计时器（1000）
我们回到nextMessage()，它返回一个可观察的但它被忽略了，因为它在上一步中被杀死了
循环回到第 4 步

一个简单的解决方法是将switchMap 更改为mergeMap，但不要忘记在内部可观察对象上添加take(1)，因为您只想为该可观察对象设置一个值：

currentMessage
  .pipe(
    tap(x => console.log('Tap:' + x)),
    mergeMap(_ => timer(1000).pipe(
      map(x => nextMessage()),
      take(1)
    ))
  )
  .subscribe(x => console.log(x));

但是，这很难理解发生了什么......如果你转向一种更具反应性的方法，我认为它会变得更简单。

您说您不能使用 interval(1000)，因为您需要根据某些条件进行重置。你为什么不把那个条件作为一个主题，这样你就可以把它写下来？看看这个：

const messageResets = new Subject();

const messages: string[] = ['Message 1', 'Message 2', 'Message 3'];
let messagePointer: number = 0;

console.clear();

messageResets.pipe(
  startWith(null),
  // Every time we need to clear the current message, we just switchMap to a new interval
  switchMap(() => timer(0, 1000).pipe(
    map(() => messages[messagePointer++ % messages.length])
  ))
).subscribe(x => console.log(x));

注意：timer(0, 1000) 与 interval(1000) 相同，但它只是在开始时发出一个值，而 interval 不会。

通常我制作主题只是为了表达流外部的东西：在这种情况下，我希望能够根据我不知道的条件从外部清除当前消息。

但是从长远来看，处理主题以保持某种内部状态总是很痛苦。

【讨论】：