RxJS中为何需在switchMap外部使用share操作符？

我在创建共享Observable时搞不懂，为什么非得在外部Observable的管道里加share操作符？我本来以为只在switchMap返回的Observable里加share就够了，但实际根本不是这么回事。

代码示例 #

const {
  catchError,
  concatMap,
  count,
  debounceTime,
  delay,
  finalize,
  map,
  mapTo,
  mergeMap,
  startWith,
  switchMap,
  tap,
  take,
  toArray,
  takeUntil,
  bufferTime,
  filter,
  bufferWhen,
  share,
  distinctUntilChanged,
} = rxjs.operators;
const {
  fromEventPattern,
  BehaviorSubject,
  Subject,
  of,
  timer,
  interval,
} = rxjs;

const first = new BehaviorSubject(1);
const destroy = new Subject();
const first$ = first.asObservable().pipe(
  map((value) => parseInt(value, 10)),
  distinctUntilChanged(),
  takeUntil(destroy)
);

const createInterval = () =>
  interval(1000).pipe(
    finalize(() => console.log('finalize')),
    share() // 此处创建多播Observable
  );

const events$ = first$.pipe(
  share(),
  switchMap(() => createInterval())
  //share() // 为何此处需要外部的share，而上方的share并不足够？
);

events$
  .pipe(tap((value) => console.log('first', value)))
  .subscribe((value) => console.log('subscription 1', value));

// 若无上方的share则无法正常订阅
events$
  .pipe(tap((value) => console.log('second', value)))
  .subscribe((value) => console.log('subscription 2', value));

HTML依赖 #

<script src="https://unpkg.com/rxjs@^7/dist/bundles/rxjs.umd.min.js"></script>

问题解析 #

要搞懂这个问题，得先明白Observable的冷/热特性，以及多次订阅时的执行逻辑：

1. 没有外部share时的问题 #

• 当你给events$添加两个订阅时，由于first$.pipe(...)是冷Observable（即使上游是BehaviorSubject，经过pipe操作符后默认还是冷的），每个新订阅都会从头执行一遍整个管道。
• 第一个订阅会触发first$的订阅，执行switchMap调用createInterval()，生成一个带share的interval流；第二个订阅会再次触发first$的订阅，再次执行switchMap生成另一个全新的interval流。
• 这时候两个interval会同时运行，你会看到两次finalize输出，两个订阅各自拿到独立的流数据，完全达不到“共享”的效果。

2. 内部share的作用局限 #

你在createInterval里加的share，只是让单个interval流本身支持多播，但问题在于：如果外部管道没有share，每次订阅events$都会触发switchMap生成一个新的带share的interval流——相当于每个订阅都有自己的独立内部流，内部的share根本没机会发挥共享作用。

3. 外部share的核心作用 #

在first$.pipe里添加share，会把整个events$变成热Observable（多播）：

• 多个订阅events$时，只会有一个订阅关联到上游的first$，switchMap只会被调用一次，生成唯一一个带share的interval流。
• 这个唯一的interval流再通过自身的share，让所有订阅共享同一份数据，最终实现所有订阅拿到相同的流，只会有一个interval在运行，finalize也只会触发一次。

总结 #

• 内部share只能让switchMap返回的单个Observable支持多播，但无法阻止外部管道因多次订阅而重复生成新的内部Observable。
• 外部share是从根源上让整个events$管道变成多播，确保所有订阅共享同一个上游执行链路，这样内部的share才能真正发挥共享流的作用。

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