rx.js 6.3中的queueScheduler是同步的 - 为什么这个例子不会导致SO如果我使用queueScheduler？

Question

我有一个有趣的例子，不是现实生活中的任务，但无论如何：

const signal = new Subject();
let count = 0;

const somecalculations = (count) => console.log('do some calculations with ', count);


console.log('Start');
signal.pipe(take(1500)/*, observeOn(queueScheduler)*/)
  .subscribe(() => {
  somecalculations(count);
  signal.next(count++);
  console.log('check if reached ', count)
});

signal.next(count++);
console.log('Stop');

codepen

Subject.next以同步的方式工作，所以如果我注释掉observeOn（queueScheduler） - 它会导致堆栈溢出（我用控制操作符控制迭代次数，而在我的计算机上如果数字大于1370则会导致SO）。

但是如果我把queueScheduler放在那里 - 它运作良好。 QueueScheduler是同步的，它以某种方式允许当前onNext处理程序运行完成运行，然后启动下一个计划运行。

有人能用源代码详细解释给我吗？我试图挖掘它，但目前取得了部分成功。这是关于observeOn如何与QueueScheduler一起工作，但答案正在逃避我。

observeOn src QueueScheduler.ts asyncScheduler

Answer 1

感谢cartant的支持。好像我理解为什么队列调度程序在没有SO的情况下工作。

1. 当第一次从observeOn _next queueScheduler.schedule-> AsyncScheduler.schedule-> Scheduler.schedule调用signal.next时会调用QueueAction.schedule
2. QueueAction.flush跟注。 this.scheduler.flush - > QueueSchedulerFlush-> AsyncScheduler.flush
3. 第一次队列为空，没有执行任务，因此this.active为false。这个action.execute的bc被称为。一切都以同步方式调用。
4. action.execute导致onNext函数再次运行。因此onNext调用signal.next它会通过所有1-3点，但现在this.active是真的（因为它实际上仍然是之前的signal.next运行）我们只是queue action
5. 所以处理第二个signal.next并返回第一个signal.next调用的action.execute。它可以在do-while和shift操作中逐个工作。所以它完成了第一个signal.next动作 - 但是现在我们还有一个来自第二个signal.next递归调用的队列。所以我们为第二个signal.next运行action.execute
6. 情况正在重演。第一次刷新调用管理所有其他调用，如：active为true，我们将任务添加到队列，然后重复上一次刷新调用并从队列中获取它。