RxJS Observable多订阅者场景下如何确保逻辑仅执行一次？

这问题我熟！你遇到的核心问题是RxJS中timer创建的是「冷Observable」，这种Observable的特性是：每有一个新订阅者，就会从头完整执行一遍整个数据流。所以你的first和second两个订阅，相当于启动了两个完全独立的timer流，每个流都会各自触发doSomething，这就导致了它被多次调用，和你预期的「不管多少订阅者，每tick只调用一次」不符。

解决思路：把冷Observable转成热Observable #

要实现「单一流执行、多订阅者共享」的效果，你需要将冷Observable转换为热Observable，RxJS提供了几个常用操作符来实现这个需求：

方案1：使用share()（最简便，推荐）

share()会自动帮你处理数据流的共享：第一个订阅者订阅时启动流，后续订阅者直接接收当前的发射值，当所有订阅者都取消订阅后，流会自动停止，完美匹配你的需求。

修改后的代码如下：

// This function shall be called *once* per tick, no matter the quantity of subscriber.
function doSomething(val) {
  console.log("doing something");
  return val;
}

// 关键：添加share()操作符实现数据流共享
const observable = Rx.Observable.timer(0, 1000)
  .map(val => doSomething(val))
  .share();

const first = observable.subscribe(val => console.log("first:", val));
const second = observable.subscribe(val => console.log("second:", val));

// After 1.5 seconds, stop first.
Rx.Observable.timer(1500).subscribe(_ => first.unsubscribe());
// After 2.5 seconds, stop second.
Rx.Observable.timer(2500).subscribe(_ => second.unsubscribe());

运行这段代码后，输出就会和你的预期完全一致：

doing something
first: 0
second: 0
doing something
first: 1
second: 1
doing something
second: 2
<nothing more>

方案2：使用publish() + connect()（更灵活）

如果你需要更精确地控制数据流的启动时机（比如要等所有订阅者都订阅完成后再开始执行），可以用publish()把Observable转换成「可连接Observable」，再手动调用connect()启动流：

// This function shall be called *once* per tick, no matter the quantity of subscriber.
function doSomething(val) {
  console.log("doing something");
  return val;
}

const observable = Rx.Observable.timer(0, 1000)
  .map(val => doSomething(val))
  .publish();

// 先添加所有订阅者
const first = observable.subscribe(val => console.log("first:", val));
const second = observable.subscribe(val => console.log("second:", val));

// 手动启动数据流
observable.connect();

// 后续取消订阅逻辑不变
Rx.Observable.timer(1500).subscribe(_ => first.unsubscribe());
Rx.Observable.timer(2500).subscribe(_ => second.unsubscribe());

这种方式的效果和share()一致，但启动时机由你手动控制，适合复杂场景。

内容的提问来源于stack exchange，提问作者FunkySayu