RxJS中Subject与Observable连接顺序差异的原因解析

我在尝试将Subject连接到interval Observable时遇到了如下问题：

代码1 #

const subject = new Subject();
    
// Connect the subject to the observable
interval(1000).subscribe(subject); //**Line 1**
 
subject.subscribe((value: any) => {
  console.log(`Subscriber A: Value: ${value}`);
});
 
setTimeout(() => {
   subject.subscribe((value: any) => {
       console.log(`Subscriber B: Value: ${value}`);
    });
}, 3000);

输出1 #

Subscriber A: Value: 0        
Subscriber A: Value: 1        
Subscriber A: Value: 2        
Subscriber A: Value: 3        
Subscriber B: Value: 3        
Subscriber A: Value: 4        
Subscriber B: Value: 4       
Subscriber A: Value: 5        
Subscriber B: Value: 5

该输出符合预期。但调整Subject与Observable的连接位置后得到代码2：

代码2 #

const subject = new Subject();
 
subject.subscribe((value: any) => {
   console.log(`Subscriber A: Value: ${value}`);
});
 
setTimeout(() => {
  subject.subscribe((value: any) => {
    console.log(`Subscriber B: Value: ${value}`);
  });
}, 3000);
    
// Connect the subject to the observable
interval(1000).subscribe(subject); //**Line 1**

输出2 #

Subscriber A: Value: 0
Subscriber A: Value: 1
Subscriber A: Value: 2
Subscriber B: Value: 2
Subscriber A: Value: 3
Subscriber B: Value: 3
Subscriber A: Value: 4
Subscriber B: Value: 4
Subscriber A: Value: 5
Subscriber B: Value: 5

疑问 #

• 为何调整Line 1的位置会导致输出差异？
• 输出2中Subscriber B从值2开始，而非像输出1那样从3开始，这是为何？我忽略了什么关键逻辑？

我原本预期代码2也会得到输出1的结果，希望了解该行为的根本原因。

解答 #

1. 调整Line1位置导致输出差异的核心原因

本质是冷Observable（interval）的启动时机和订阅关系建立顺序共同作用：

• interval是冷Observable，只有被订阅时才会启动计时并发射值。
• 代码1中，先让interval订阅Subject，此时interval已开始计时，但Subject还没有下游订阅者，interval的发射不会停止，只是值暂时无人接收；直到Subscriber A订阅Subject后，后续所有值都会被转发给A。
• 代码2中，先让Subscriber A订阅Subject，再让interval订阅Subject，此时interval启动时Subject已有订阅者，所以从第一个值0开始就会被同步转发给A。

2. Subscriber B在输出2中从值2开始的原因

这是宏任务队列的执行顺序和Subject的特性共同导致的：

• 代码2中，setTimeout的3秒回调和interval发射值2的回调，会在代码执行后的第3000ms左右同时进入宏任务队列。假设setTimeout回调先执行：此时Subscriber B完成订阅，成为Subject的下游订阅者；紧接着interval的发射回调执行，Subject会将值2同步推送给所有当前订阅者（A和刚订阅的B），因此B会立刻收到值2。
• 代码1中，interval在第3000ms时已经完成值2的发射（推送给A），之后setTimeout回调才执行，B订阅时没有当前值可接收，只能等待下一个值3的发射，所以B从3开始输出。

另外要注意：普通Subject不会缓存任何历史值，只会转发订阅之后的新值，这也是B无法收到订阅前的值的原因。

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