system.reactive - 为什么 Rx Observable.Subscribe 会阻塞我的线程？

Question

您好，我已经尝试了 101 个 Rx 示例之一:

    static IEnumerable<int> GenerateAlternatingFastAndSlowEvents()
    {
        int i = 0;

        while (true)
        {
            if (i > 1000)
            {
                yield break;
            }
            yield return i;
            Thread.Sleep(i++ % 10 < 5 ? 500 : 1000);
        }
    }

    private static void Main()
    {
        var observable = GenerateAlternatingFastAndSlowEvents().ToObservable().Timestamp();
        var throttled = observable.Throttle(TimeSpan.FromMilliseconds(750));

        using (throttled.Subscribe(x => Console.WriteLine("{0}: {1}", x.Value, x.Timestamp)))
        {
            Console.WriteLine("Press any key to unsubscribe");
            Console.ReadKey();
        }

        Console.WriteLine("Press any key to exit");
        Console.ReadKey();
    }

我不明白为什么“按任意键取消订阅”这一行从未显示。我的理解是订阅是异步的，你订阅它会立即返回。我错过了什么导致我的主线程阻塞？

Answer 1

阻塞是由您在 while (true) 上的可枚举循环的组合引起的。和 IEnumerable<T>.ToObservable()扩展方法默认为 CurrentThreadScheduler .

如果您供应 Scheduler.TaskPool (或 Scheduler.ThreadPool 在 pre-.NET 4 中)到 ToObservable 的过载，您应该会看到您期望的行为(尽管它不会在主线程上调用您的订阅者，仅供引用)。

话虽如此，我想你会找到你的组合Thread.Sleep和 Throttle将按您的预期工作。您可能最好创建一个使用调度程序来安排延迟的自定义 observable。