当我将延迟运算符添加到第二个订阅时，共享一个 observable 变为单播

Question

以下代码按预期工作：

    const source = interval(1000).pipe(
        take(5),    
        share()
  );

  source.subscribe(x => console.log('c1', x));
  setTimeout(() => {
    source.subscribe(x => console.log('c2', x));
  }, 2000);

产生以下输出： c1 0 c1 1 c1 2 c2 2 c1 3 c2 3 c1 4 c2 4

但是当我将第二个订阅更改为使用 delay(2000) 而不是 setTimeout() 我收到了另一个未共享的流。

    const source = interval(1000).pipe(
        take(5),    
        share()
  );

  source.subscribe(x => console.log('c1', x));

  source.pipe(delay(2000)).subscribe(x => console.log('c2', x));

产生这个输出：

c1 0 c1 1 c1 2 c2 0 c1 3 c2 1 c1 4 c2 2 c2 3 c2 4

如何让第二个订阅者使用共享流？ 我显然不完全了解 RX 操作员是如何在幕后工作的。

Answer 1

使用source.pipe(delay(2000)) 与使用setTimeout() 完全不同。 delay() 运营商将延迟其源的每次发射，这意味着您仍会立即进行两次订阅。

您可能想要做的是：

of(null)
  .pipe(
    delay(2000),
    switchMapTo(source),
  )
  .subscribe();

或者这应该做同样的事情：

concat(timer(2000), source)
  .subscribe();

Answer 2

视觉效果

这是 c2 两秒后订阅的原始流

src: 0-1-2-3-4
 c1: 0-1-2-3-4
 c2: ----2-3-4

以及 c2 立即订阅并将每次发射延迟 2 秒的流

src: 0-1-2-3-4
 c1: 0-1-2-3-4
 c2: ----0-1-2-3-4

分享来源

共享源不等于拥有相同的输出。

考虑一下：

const src = interval(1000).pipe(
  take(5),    
  map(x => x + 5),
  share()    
);

src.pipe(map(x => x - 1)).subscribe(console.log); // c1
src.pipe(map(x => x + 1)).subscribe(console.log); // c2

输出：

src: 5-6-7-8-9
 c1: 4-5-6-7-8
 c2: 6-7-8-9-10

即使它们都有不同的输出。 c1 和 c2 都具有相同的来源。 c1 和 c2 不生成任何数字，它们只是从给定的任何数字中加 1 并减 1。他们改变他们的来源。

这与您在第二个示例中所做的相同。当发射发生时延迟会改变，而不是转换数字。 c2 发出与源相同的流，它在 2 秒后才开始发出，在源完成后 2 秒仍在发出。

用 setTimeout 延迟

const source = interval(1000).pipe(
  take(5),    
  share()
);

source.subscribe(x => console.log('c1', x));
source.subscribe(x =>
  setTimeout(_ => 
    console.log('c2', x), 
    2000
  )
);

订阅前等待

const source = interval(1000).pipe(
  take(5),    
  share()
);
source.subscribe(x => console.log('c1', x));
timer(2000).pipe(
  switchMap(_ => source)
).subscribe(x => console.log('c2', x));