多播场景下Subject为何无法接收Observable的complete()信号？

问题原因 #

1. Observable构造函数返回的退订清理函数，执行时机是订阅已经被取消之后，此时关联的Subscriber已经处于关闭状态，调用它的complete()、error()、next()方法都会被静默忽略，通知无法下发到下游的Subject和观察者，因此你定义的完成提示不会触发。
2. 代码存在拼写错误：subject.subsribe(observer) 漏写了字母c，正确写法为subject.subscribe(observer)，即便修正该拼写问题，核心逻辑问题依然存在。
3. 调用connectableObservable.connect()返回的订阅是上游Observable到Subject的链路订阅，调用unsubscribe()只会取消该链路的订阅，不会主动给Subject发送完成通知，下游观察者自然收不到对应信号。

error和complete的正确使用规则 #

• 调用时机限制：complete()、error()必须在订阅处于激活状态时调用，也就是要放在Observable构造函数的同步/异步执行逻辑里，禁止放到退订清理函数中执行。如果希望上游发完指定值后自动触发完成，直接在最后一个next()调用后执行complete()即可。
• 终止互斥特性：同一个Subscriber触发complete()或者error()任意一个终止通知后，后续的所有next()、以及另一种终止通知都会被自动忽略，不会下发给观察者。
• 多播场景传播规则：Subject作为多播载体，收到上游的complete()/error()通知后会进入终止状态，后续新订阅该Subject的观察者，会直接收到对应的终止通知。
• 退订与终止的区别：主动调用unsubscribe()只是取消订阅链路，不会触发任何终止通知。如果需要退订时同时下发完成信号，需要在调用unsubscribe()前手动给Subject调用complete()。
• 退订清理函数的定位：该函数的作用是释放订阅占用的资源（比如清理定时器、关闭网络连接等），不应该用来下发任何通知。

修正后代码示例 #

import { Observable, Subject, connectable } from 'rxjs';

// 方式1：上游发完值后自动触发complete
let source = new Observable(subscriber => {
   subscriber.next(1);
   subscriber.next(2);
   // 订阅激活状态下调用complete，通知可正常下发
   subscriber.complete();
   return () => { 
      console.log('Unsubscribed');
   }
});

let subject = new Subject<number>();
const observer = {
  next: (x) => console.log('Observer got a next value: ', x),
  error: (err) => console.error('Observer got an error: ', err),
  complete: () => console.log('Observer got a complete notification'),
};
subject.subscribe(observer);
const connectableObservable = connectable(source, {
  connector: () => subject,
});
const subscription = connectableObservable.connect();

// 如果需要主动触发完成再退订，用下面的写法
// subject.complete();
// subscription.unsubscribe();

内容的提问来源于stack exchange，提问作者abigail.nguyen