rxjs - `observeOn`和参数化调度程序之间的区别

Question

我希望以下两段代码是等效的。默认情况下，Repeat 使用 currentThread 调度程序。如果我们将其更改为立即调度程序:

Rx.Observable.fromArray([1,2,3,4,5]).flatMap(a => {
  return Rx.Observable.repeat(a, 3, Rx.Scheduler.immediate)
})
.subscribe(r => console.log(r));

上面的代码产生预期的结果:1, 1, 1, 2, 2, 2, ...。但以下代码不会，并会生成一系列混合值:

Rx.Observable.fromArray([1,2,3,4,5]).flatMap(a => {
  return Rx.Observable.repeat(a, 3).observeOn(Rx.Scheduler.immediate)
})
.subscribe(r => console.log(r));

我不明白这种行为，但我想我错过了一些东西。 repeat 可以传递一个 Scheduler 参数，但我想我也可以通过使用 observeOn 来强制 Observable 作用于特定的 Scheduler。我错过了什么？

Answer 1

区别在于，第一个使用调度程序进行生成，第二个仅使用它进行传播。

在第二个版本中，您仍然使用currentThread来创建值。 observeOn 只会在值从前一个运算符发出后将值强制到不同的调度程序上，但对于生成事件的运算符，这不会影响这些事件的生成。

如果您查看某些创建运算符(例如 fromArray)的内部，您会看到类似以下内容的内容:

//Changing the scheduler will change how recursive scheduling works
scheduler.schedulerRecursiveWithState(0, function(self, state) {
      if (i < len) {
        observer.onNext(array[i]);
        //Schedule the next event
        self(i + 1);
      } else {
        observer.onCompleted();
      }
});

而 observeOn 类似于执行以下操作:

//Doesn't change when events get generated, simply reschedules them for down stream
source.subscribe(function(x) {
  scheduler.scheduleWithState(x, function(self, state) {
    observer.onNext(x);
  });
});