调度程序:立即与当前线程 [英] Schedulers: Immediate vs. CurrentThread

问题描述

阅读解释为什么

Observable.Return(5)
  .Repeat()
  .Take(1)

永远不会完成，但是

Observable.Return(5, Scheduler.CurrentThread)
  .Repeat()
  .Take(1)

按预期工作.我仍然很困惑，我不知道为什么 currentThread 实际上解决了这个问题.谁能给个明确的解释?

works as expected. I am still confused and I can't tell why currentThread actually solves the problem. Can somebody give a clear explanation?

推荐答案

link 由 Ned Stoyanov 在上面的评论中提供，Dave Sexton 有很好的解释.

The link provided by Ned Stoyanov in the comments above has a great explanation by Dave Sexton.

我会尝试用不同的方式来说明.以 RecursiveMethod 中发生递归调用为例.

I'll try to illustrate it a bit differently. Take this example where a recursive call occurs in the RecursiveMethod.

public class RecursiveTest()
{
    private bool _isDone;

    public void RecursiveMethod()
    {
        if (!_isDone)
        {
            RecursiveMethod();

           // Never gets here...
           _isDone = true;
        }
    }  
}

您可以很容易地看到这将无限递归(直到出现 StackOverflowException)，因为 _isDone 永远不会被设置为 true.这是一个过于简化的示例，但基本上就是第一个示例的情况.

You can easily see that this will recurse indefinitely (until a StackOverflowException) because _isDone will never gets set to true. It is an overly simplified example, but it is basically what's going on with your first example.

这是 Dave Sexton 的解释，用于描述您的第一个示例中发生的情况.

This is the explanation by Dave Sexton to describe what happens in your first example.

默认情况下，Return 使用 ImmediateScheduler 调用 OnNext(1) 然后OnCompleted().重复不引入任何并发，所以它看到OnCompleted 立即然后立即重新订阅 Return.因为回归中没有蹦床，所以这种模式会重复，无限期地阻塞当前线程.在此调用订阅observable 永远不会返回.

By default, Return uses the ImmediateScheduler to call OnNext(1) then OnCompleted(). Repeat does not introduce any concurrency, so it sees OnCompleted immediately and then immediately resubscribes to Return. Because there's no trampoline in Return, this pattern repeats itself, blocking the current thread indefinitely. Calling Subscribe on this observable never returns.

换句话说，由于可重入的无限循环，初始流程永远不会完全完成.所以我们需要一种方法来完成初始流程而无需重入.

In other words, because of the infinite loop of reentrancy, the initial flow never gets fully completed. So we need a way to complete the initial flow without the reentrancy.

让我们回到本文上面的 RecursiveTest 示例，避免无限递归的解决方案是什么?在再次执行 RecursiveMethod 之前，我们需要 RecursiveMethod 完成其流程.一种方法是创建一个队列并将对 RecursiveMethod 的调用加入队列，如下所示:

Let's go back to my RecursiveTest example above in this post, what would be the solution to avoid infinite recursion? We would need the RecursiveMethod to complete its flow before executing again the RecursiveMethod. One way to do this is to have a queue and enqueue the call to the RecursiveMethod like this:

public void RecursiveMethod()
{
    if (!_isDone)
    {
        Enqueue(RecursiveMethod);
        _isDone = true;
    }
}  

这样，初始流程将完成，_isDone 将被设置为 true，并且在执行下一次对 RecursiveMethod 的调用时，将不再执行任何操作，避免了无限递归.这几乎就是 Scheduler.CurrentThread 将对您的第二个示例执行的操作.

This way, the initial flow would complete, _isDone would be set to true and when the next call to RecursiveMethod is executed, nothing will get executed anymore avoiding the infinite recursion. And this is pretty much what the Scheduler.CurrentThread will do to your second example.

让我们看看 Dave Sexton 如何解释您的第二个示例的工作原理:

Let's see how Dave Sexton explains how your second example works:

这里，Return 是使用 CurrentTheadScheduler 调用 OnNext(1) 然后OnCompleted().重复不引入任何并发，所以它看到OnCompleted 立即完成，然后立即重新订阅 Return；然而，这第二个对 Return 的订阅安排了它的(内部)在蹦床上的动作，因为它仍在执行来自第一个计划(外部)操作的 OnCompleted 回调，因此重复不会立即发生.这允许重复返回一个对 Take 来说是一次性的，它最终调用 OnCompleted，取消了通过处理Repeat来重复，最终是来自Subscribe的调用返回.

Here, Return is using the CurrentTheadScheduler to call OnNext(1) then OnCompleted(). Repeat does not introduce any concurrency, so it sees OnCompleted immediately and then immediately resubscribes to Return; however, this second subscription to Return schedules its (inner) actions on the trampoline because it's still executing on the OnCompleted callback from the first scheduled (outer) action, thus the repetition does not occur immediately. This allows Repeat to return a disposable to Take, which eventually calls OnCompleted, cancels the repetition by disposing Repeat, and ultimately the call from Subscribe returns.

同样，我的示例确实经过简化以使其易于理解，但它并不完全是如何工作的.可以看到调度程序是如何工作的.它使用他们所谓的 Trampoline，它基本上是一个队列，可确保没有可重入调用.因此，所有调用都在同一线程上一个接一个地序列化.这样就可以完成初始流程，避免了无限重入循环.

Again my example was really simplified to make it easy to understand and it's not exactly how it works. Here you can see how the scheduler really works. It uses what they call a Trampoline which is basically a queue that makes sure that there is no reentrant calls. All calls are therefore serialized one after the other on the same thread. By doing so, the initial flow can be completed which avoids the infinite reentrant loop.

希望这更清楚一点:)

这篇关于调度程序:立即与当前线程的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！