Rx代码阻止UI线程？

Question

我正在体验反应式编程（Rx），其中一个有趣的功能是订阅和观察不同的线程。但这里以某种方式阻止了UI线程。从技术上讲，我没有任何方法返回Task（异步方法），所以在这里我试图用Thread.Sleep模仿一个漫长的过程：

IEnumerable<Item> _search(string searchText)
{
        Thread.Sleep(3000);
        //return result by querying ...
        //...
        return someResult;
}

我有一个像这样的ViewModel类：

public class ViewModel {
   public ViewModel(){
       //this SubscribeOn may not be necessary but I just try it here for sure
       SearchTextStream.SubscribeOn(NewThreadScheduler.Default)
                       .ObserveOn(DispatcherScheduler.Current)
                       .Subscribe(searchText => {
                          var items = _search(searchText);
                 }, ex => {
                    //handle error
            });
   }
   public string SearchText
   {
        get
        {
            return _searchText.FirstAsync().Wait();
        }
        set
        {
            _searchText.OnNext(value);
        }
   }
   ISubject<string> _searchText = new BehaviorSubject<string>("");
   public IObservable<string> SearchTextStream
   {
        get
        {
            return _searchText.AsObservable().DistinctUntilChanged();
        }
   }
}

实际上没有使用Thread.Sleep，我仍然可以看到它阻止UI但不是很明显，所以我只是用它来使它更明显。正如我所说，这里的情况是我只有一个没有任何任务或异步的普通方法。这可能是一个长期运行的方法。使用RX，我不知道应该做些什么来使它像async一样（如使用Task.Run时）？

如果重要，我正在测试WPF应用程序。

Answer 1

你在_search(searchText)调度程序上调用DispatcherScheduler.Current - 因此，使用Thread.Sleep你会阻止UI。

你真的应该让_search返回一个可观察的。

IObservable<IEnumerable<Item>> _search(string searchText)
{
    Thread.Sleep(3000);
    //return result by querying ...
    //...
    return Observable.Return(new [] { new Item() });
}

现在构造函数应该如下所示：

public ViewModel()
{
    SearchTextStream
        .ObserveOn(System.Reactive.Concurrency.Scheduler.Default)
        .SelectMany(searchText => _search(searchText))
        .ObserveOnDispatcher()
        .Subscribe(items =>
        {
            /* do something with `items` */
        }, ex =>
        {
            //handle error
        });
}