RxJS：如何让Observable多次订阅时subscribe回调仅执行一次？

两种核心解决方案 #

1. 只接收Observable的第一次发射值 #

如果业务仅需要获取Observable的第一个值，后续无需监听新值，可以用first()或take(1)操作符。它们会在Observable发射第一个值后自动完成订阅，后续新值不会触发回调：

this.store.select(MsnSelectors.getUserWhoTyping)
  .pipe(first()) // 也可以用 take(1)，效果一致
  .subscribe(uts => {
    if (uts.length > 0) {
      const ut = uts.find(x => x.senderUserName === this.receiver.identifier);
      console.log(ut);
      this.userTyping = ut?.isTyping ? ut : null;
    } else {
      this.userTyping = null;
    }
  });

2. 持续监听但仅保留一个订阅实例 #

如果需要持续接收Observable的新值，但不想因为重复执行订阅代码而创建多个订阅（导致回调重复触发），可以用以下两种方式：

方式A：复用订阅对象，避免重复订阅

在组件类中保存订阅实例，每次订阅前先取消之前的订阅（记得在组件销毁时清理，防止内存泄漏）：

import { Subscription } from 'rxjs';

// 组件类中定义订阅变量
private typingSubscription: Subscription | null = null;

// 封装订阅逻辑
setupTypingListener() {
  // 先取消已存在的订阅
  this.typingSubscription?.unsubscribe();
  
  this.typingSubscription = this.store.select(MsnSelectors.getUserWhoTyping)
    .subscribe(uts => {
      if (uts.length > 0) {
        const ut = uts.find(x => x.senderUserName === this.receiver.identifier);
        console.log(ut);
        this.userTyping = ut?.isTyping ? ut : null;
      } else {
        this.userTyping = null;
      }
    });
}

// Angular组件销毁时清理订阅
ngOnDestroy() {
  this.typingSubscription?.unsubscribe();
}

方式B：将Observable转为多播模式

用shareReplay(1)操作符让Observable变为多播，所有订阅者共享同一个底层订阅，这样即使多次调用subscribe，也只会触发一次回调逻辑：

import { shareReplay } from 'rxjs/operators';

// 在组件初始化时处理Observable（比如构造函数或ngOnInit）
private userWhoTyping$ = this.store.select(MsnSelectors.getUserWhoTyping)
  .pipe(shareReplay(1));

// 后续多次订阅这个处理后的Observable即可
this.userWhoTyping$.subscribe(uts => {
  if (uts.length > 0) {
    const ut = uts.find(x => x.senderUserName === this.receiver.identifier);
    console.log(ut);
    this.userTyping = ut?.isTyping ? ut : null;
  } else {
    this.userTyping = null;
  }
});

内容的提问来源于stack exchange，提问作者adnen ben awicha