Rx.net - 同步与异步观察者 - 取决于来源？

Question

我对 Rx 非常陌生，并试图围绕它进行思考。没有阅读很多内容，但尝试先通过动手实验。

class Program
{
    static void Main(string[] args)
    {
        // one source, produces values with delays
        IObservable<int> source = Observable.Generate(0, i => i < 2, i => ++i, i => i*i, i => TimeSpan.FromMilliseconds(100));
        IObserver<int> handler = null;

        IDisposable subscription = source.Subscribe(
            i =>
            {
                Console.WriteLine("Sub 1 [tid:{0}] received {1} from source", Thread.CurrentThread.ManagedThreadId,i);
                Thread.Sleep(500);
            },
            exception => Console.WriteLine("Sub 1 Something went wrong {0}", exception),
            () => Console.WriteLine("Sub 1 Completed observation"));

        IDisposable subscription2 = source.Subscribe(
            i => Console.WriteLine("Sub 2 [tid:{0}] received {1} from source", Thread.CurrentThread.ManagedThreadId, i),
            exception => Console.WriteLine("Sub 2 Something went wrong {0}", exception),
            () => Console.WriteLine("Sub 2 Completed observation"));

        Console.WriteLine("press to cancel");
        Console.ReadLine();
        subscription.Dispose();
        subscription2.Dispose();

    }
}

这会按预期产生异步交错执行。

另一方面，如果我将源更改为同步，即使观察者也将变为阻塞和同步（相同的线程 id，在没有完全消耗 sub1 的情况下不会转到 sub2）。 有人可以帮我理解这一点吗？这是同步版本

class Program
{
    static void Main(string[] args)
    {
        // one source, produces values
        IObservable<int> source = Observable.Generate(0, i => i < 2, i => ++i, i => i*i);
        IObserver<int> handler = null;

        // two observers that consume - first with a delay and the second immediately.
        // in this case, the behavior of the observers becomes synchronous? 
        IDisposable subscription = source.Subscribe(
            i =>
            {
                Console.WriteLine("Sub 1 [tid:{0}] received {1} from source", Thread.CurrentThread.ManagedThreadId,i);
                Thread.Sleep(500);
            },
            exception => Console.WriteLine("Sub 1 Something went wrong {0}", exception),
            () => Console.WriteLine("Sub 1 Completed observation"));

        IDisposable subscription2 = source.Subscribe(
            i => Console.WriteLine("Sub 2 [tid:{0}] received {1} from source", Thread.CurrentThread.ManagedThreadId, i),
            exception => Console.WriteLine("Sub 2 Something went wrong {0}", exception),
            () => Console.WriteLine("Sub 2 Completed observation"));

        Console.WriteLine("press to cancel");
        Console.ReadLine();
        subscription.Dispose();
        subscription2.Dispose();

    }
}

Answer 1

我相信原因是为运营商选择了默认IScheduler。看看接受的答案here。

对于Generate，它取决于过载。根据答案，这些是使用的默认调度程序。如果您愿意，可以验证它们的来源

• 时间运算符的默认IScheduler 是DefaultScheduler.Instance
• 后一个运算符的默认IScheduler 是CurrentThreadScheduler.Instance

您可以通过在同步版本中提供“非阻塞”调度程序来确认这一点

IObservable<int> source = Observable.Generate(0, i => i < 2, i => ++i, i => i * i, DefaultScheduler.Instance);