为什么 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) 上的可枚举循环和默认为 CurrentThreadScheduler 的 IEnumerable.ToObservable() 扩展方法的组合引起的>。

如果您向 ToObservable 的重载提供 Scheduler.TaskPool（或 .NET 4 之前的 Scheduler.ThreadPool），您应该会看到以下行为您正在期待（尽管它不会在主线程上调用您的订阅者，仅供参考）。

话虽如此，我认为您会发现 Thread.Sleep 和 Throttle 的组合将按您的预期工作。您可能最好创建一个使用调度程序来安排延迟的自定义可观察对象。

Answer 2

我同意理查德的观点。

.ToObservable() 的实现如下所示：

public static IObservable<TSource> ToObservable<TSource>(
    this IEnumerable<TSource> source)
{
    if (source == null)
    {
        throw new ArgumentNullException("source");
    }
    return source.ToObservable<TSource>(Scheduler.CurrentThread);
}

它使用 Scheduler.CurrentThread 调用 .ToObservable(IScheduler) 重载，因为您使用 .Sleep(...) 导致可观察对象必须完成延迟，然后代码才能超越 .Subscribe(...) 方法。只要想想如果这段代码全部在单个线程上运行（事实确实如此），它会是什么样子。

要解决这个问题，您可以像理查德建议的那样使用任务池或线程池调度程序，但我认为您有更多您的代码的根本问题。也就是说，您正在使用“老派”线程休眠，而不是依赖 Rx 方法。

尝试生成您的可观察值：

var observable =
    Observable
        .GenerateWithTime(0, i => i <= 1000, i => i + 1,
            i => i, i => TimeSpan.FromMilliseconds(i % 10 < 5 ? 500 : 1000))
        .Timestamp();

GenerateWithTime(...) 执行您的 GenerateAlternatingFastAndSlowEvents 方法所做的所有操作，但它直接创建可观察值并使用 Scheduler 执行此操作.ThreadPool 在底层，因此您不需要指定任何调度程序。