Rx 2.1：如何正确订阅和观察 Dispatcher答案

【问题标题】：Rx 2.1: How to subscribe and observe on Dispatcher correctlyRx 2.1：如何正确订阅和观察 Dispatcher
【发布时间】：2013-02-28 20:27:54
【问题描述】：

首先，我使用的是最新的 Rx，即 2.1。据我了解，当 Rx 2 岁时，很多事情都发生了变化，所以我真的很期待收到最新的答案。提前致谢。

我正在为 Rx 实现一个经典任务：观察 TextBox 的文本（确切地说是来自 WPToolkit 的 AutoCompleteBox），以便向用户提供建议列表。建议是从网络获取的，我想使用这些普通的 Rx 好东西，比如 Throttle、DistinctUntilChanged 等。

我也在使用最近发布的适用于 Windows Phone 8 的便携式 HttpClient，因为它提供了基于任务的异步 API，这很好。

我遇到的问题是在读取“AutoCompleteBox”的Text 值时的跨线程访问。代码如下：

var http = new HttpClient();
var searchFunc = Observable.FromAsync<HttpResponseMessage>(() => 
            http.GetAsync(FormatUrl(SEARCH_URL, "DE", new GeoCoordinate(51, 13), searchBox.Text /* <-- causes exception */, 10, "")));

var uithread = new SynchronizationContextScheduler(SynchronizationContext.Current);
var textChange = Observable.FromEventPattern<RoutedEventArgs>(searchBox, "TextChanged")                             
        .Throttle(TimeSpan.FromMilliseconds(800))
        .DistinctUntilChanged()     
        .SubscribeOn(uithread)           
        .SelectMany(searchFunc)                
        .Select(async (resp) => SearchResultsParser.ParseSearchResults(await resp.Content.ReadAsStreamAsync(), new GeoCoordinate(51, 13)))
        .Select(async (results) => searchBox.ItemsSource = await results)
        .ObserveOn(uithread)
        .Subscribe();

执行searchFunc 时发生异常。尽管我使用了 SubscribeOn，但我从 VS 中看到它在工作线程上执行。

这是使用SynchronizationContextScheduler 的示例，但我也尝试了SubscribeOnDispatcher，结果相同。看起来我在这个ObserveOn 的东西上遗漏了一些重要的东西，或者可能是关于Observable.FromAsync。你能指出我的错误吗？

【问题讨论】：

你粘贴的那个代码，它是从哪里执行的？在表单/控件内？（基本上，“它确定在 ui 线程上执行吗？”）
是的，它确实是在 UI 线程上执行的。这是页面的构造函数。

标签： windows-phone-8 system.reactive observable

【解决方案1】：

SubscribeOn几乎永远不是你想要的 - 你可能认为它的意思是“我的 Subscribe 方法运行的地方”，但它实际上意味着“到 IDisposable 的实际接线位置（和处置）运行”-ObserveOn 相当于“这是我希望我的实际 Subscribe 代码执行的地方”

参考：Observable.SubscribeOn and Observable.ObserveOn

【讨论】：

不幸的命名，那个。