RxJs 共享操作符和使用范围创建的 Observable

Question

如果源 Observable 是使用 range 而不是 timer 创建的，我试图理解为什么 share RxJs 运算符的工作方式不同。

将原代码改为：

const source = range(1, 1)
    .pipe(
        share()
    )

const example = source.pipe(
    tap(() => console.log('***SIDE EFFECT***')),
    mapTo('***RESULT***'),
)

const sharedExample = example
const subscribeThree = sharedExample.subscribe(val => console.log(val))
const subscribeFour = sharedExample.subscribe(val => console.log(val))

结果：

console.log src/pipeline/foo.spec.ts:223 副作用

console.log src/pipeline/foo.spec.ts:228 结果

console.log src/pipeline/foo.spec.ts:223 副作用

console.log src/pipeline/foo.spec.ts:229 结果

基本上，副作用会被多次调用。

据我所知，range 应该是一个冷的 observable，但据说 share 应该把冷的 observable 变成热的。

这种行为背后的解释是什么？

Answer 1

有两点需要指出。

首先，如果您仔细查看 range 的函数签名，您会发现它需要第三个参数 SchedulerLike。

如果未指定，RxJS立即调用每个订阅者的 next 处理程序，并使用 range 可观察对象的相关值，直到它耗尽。如果您打算使用 share 运算符，这是不可取的，因为它有效地绕过了可能引入的任何共享副作用处理。

Relevant snippet taken from the actual implementation:

// src/internal/observable/range.ts#L53
do {
  if (index++ >= count) {
    subscriber.complete();
    break;
  }
  subscriber.next(current++);
  if (subscriber.closed) {
    break;
  }
} while (true);

timer 还接受一个可选的SchedulerLike 参数。如果不指定，实现默认采用AsyncScheduler，不同于range的默认实现。

其次，share 运算符应遵循所有其他可能有副作用的运算符。如果它在它们之前，则管道运算符处理的预期统一行为将丢失。

因此，考虑到这两点，让share 运算符与range 一起工作，就像您所期望的那样：

const { asyncScheduler, range, timer } = rxjs;
const { mapTo, tap, share } = rxjs.operators;

// Pass in an `AsyncScheduler` to prevent immediate `next` handler calls
const source = range(1, 1, asyncScheduler).pipe(
  tap(() => console.log('***SIDE EFFECT***')),
  mapTo('***RESULT***'),
  // All preceding operators will be in shared processing
  share(),
);

const sub3 = source.subscribe(console.log);
const sub4 = source.subscribe(console.log);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.4.0/rxjs.umd.min.js"></script>