RxJS:延迟的内部可观察对象仍然运行，即使外部可观察对象没有剩余的订阅

Question

我刚刚意识到，内部可观察对象(就像在mergeMap或switchMap操作符中定义的那些)即使外部可观察对象没有剩余的订阅，也不会“停止”。

为了获得更好的示例，让我们展示一些代码：

const {
  Subject,
  of: obsOf,
  concat: obsConcat,
  defer,
} = require("rxjs");
const {
  finalize,
  mergeMap,
  tap,
  takeUntil,
} = require("rxjs/operators");

const subject = new Subject();

obsOf(null).pipe(
  mergeMap(() =>
    obsConcat(
      defer(() => {
        console.log("side-effect 1");
        return obsOf(1);
      }),
      defer(() => {
        console.log("side-effect 2");
        return obsOf(2);
      }),
      defer(() => {
        console.log("side-effect 3");
        return obsOf(3);
      })
    )
  ),
  finalize(() => {
    console.log("finalized");
  })
)
.pipe(
  takeUntil(subject),
  tap((i) => {
    if (i === 2) {
      subject.next();
    }
  })
).subscribe(
  (i) => { console.log("next", i); },
  (e) => { console.log("error", e); },
  () => { console.log("complete"); },
);

// Ouput:
// > side-effect 1
// > next 1
// > side-effect 2
// > complete
// > finalized
// > side-effect 3

由于外部观察值已经被称为finalize，因此记录side-effect 3行的事实很奇怪。

因为所有这些副作用都在defer中，所以在取消订阅后，它们可以完美地避免。在我看来，这些副作用根本没有价值。

知道为什么RxJS还在执行这些代码吗？

Answer 1

不幸的是，这是设计上的(从RxJS 6开始)- concat将缓冲可观察对象，并将订阅每个缓冲的对象，即使在您取消订阅之后(如果订阅是closed，它将订阅并立即取消订阅)。

你必须防止可观察到的东西被缓冲...

obsOf(null).pipe(
  mergeMap(() => obsOf(
    defer(() => {
      console.log("side-effect 1");
      return obsOf(1);
    }),
    defer(() => {
      console.log("side-effect 2");
      return obsOf(2);
    }),
    defer(() => {
      console.log("side-effect 3");
      return obsOf(3);
    })
  )),
  concatAll(),
  finalize(() => {
    console.log("finalized");
  }),
  takeUntil(subject),
  tap((i) => {
    if (i === 2) {
      subject.next();
    }
  })
).subscribe(
  (i) => { console.log("next", i); },
  (e) => { console.log("error", e); },
  () => { console.log("complete"); },
);

人们可能会认为上面的代码是有效的，但直到你延迟了其中一个可观察到的代码。用timer(100).pipe(mapTo(1));替换obsOf(1)和行为完全相同。

唯一的解决办法是确保您没有缓冲任何内容(意思是不使用concat*操作符)，或者以其他方式限制可观察到的生产(使用单独的主体并手动控制生产)。