缓冲数据库游标中的数据，同时保持 UI 响应

我有一个已填充的数据库目录和一个可用于检索对象的游标。这个目录显然可能非常大，我想做的是使用反应式用户界面缓冲数据，同时保持 UI 数据绑定和响应。我按照步骤操作here http://www.interact-sw.co.uk/iangblog/2013/02/20/wpf-rx-threads-chunking翻译我的IEnumerable进入一个IObservable，如下所示：

public class CatalogService
{
   ...

   public IObservable<DbObject> DataSource
   {
        get
        {
            return Observable.Create<DbObject>(obs =>
            {
                var cursor = Database.Instance.GetAllObjects();
                var status = cursor.MoveToFirst();

                while (status == DbStatus.OK)
                {
                    var dbObject= Db.Create(cursor);
                    obs.OnNext(dbObject);

                    status = cursor.MoveToNext();
                }

                obs.OnCompleted();

                return Disposable.Empty;
            });
        }
    }
}

在我看来，班级（具体来说，Loaded事件），我正在订阅数据源并使用缓冲区方法，希望保持 UI 响应。

    public ObservableCollection<DbObject> DbObjects { get; set; }

    private async void OnLoad(object sender, RoutedEventArgs e)
    {
        var observableData = CatalogService.Instance.DataSource.Publish();
        var chunked = observableData.Buffer(TimeSpan.FromMilliseconds(100));
        var dispatcherObs = chunked.ObserveOnDispatcher(DispatcherPriority.Background);
        dispatcherObs.Subscribe(dbObjects =>
        {
            foreach (var dbObject in dbObjects)
            {
                DbObjects.Add(dbObject);
            }
        });

        await Task.Run(() => observableData.Connect());
        await dispatcherObs.ToTask();
    }

不幸的是，结果恰恰相反。当我的视图控件（其中包含一个简单的ListBox数据绑定到DbObjects属性）加载，在枚举整个目录之前它不会显示任何数据。只有这样，UI 才会刷新。

我是 ReactiveUI 的新手，但我确信它能够完成手头的任务。如果我使用不正确，有人有任何建议或指示吗？

在等待更多信息之前，我的猜测是，您可能有几个零长度缓冲区，具体取决于数据库查询所需的时间，然后是一个包含所有结果的非零长度缓冲区。您最好通过长度和时间来限制缓冲区大小。

编辑-我只是想对原始实现中涉及的各种线程进行分析。我不同意Paul的分析，我不相信UI线程因为数据库查询而被阻塞。我相信它被阻止是由于大量结果被缓冲。

查理 - 拜托，你能在代码中（而不是使用调试器）计时数据库查询并转储你得到的缓冲区长度吗？

我将注释代码以显示所涉及的所有三个线程的顺序：

首先，在提供的代码之外，我假设调用了OnLoad通过Loaded event.

(1) - UI 线程调用 OnLoad

public ObservableCollection<DbObject> DbObjects { get; set; }

private async void OnLoad(object sender, RoutedEventArgs e)
{
    // (2) UI Thread enters OnLoad

    var observableData = CatalogService.Instance.DataSource.Publish();

    var chunked = observableData
        // (6) Thread A OnNext passes into Buffer
        .Buffer(TimeSpan.FromMilliseconds(100));
        // (7) Thread B, threadpool thread used by Buffer to run timer 

    var dispatcherObs = chunked
        // (8) Thread B still
        .ObserveOnDispatcher(DispatcherPriority.Background);
        // (9) Non blocking OnNexts back to UI Thread

    dispatcherObs.Subscribe(dbObjects =>
    {
        // (10) UI Thread receives buffered dbObjects            
        foreach (var dbObject in dbObjects)
        {
            // (11) UI Thread hurting while all these images are
            // stuffed in the collection in one go - This is the issue I bet.
            DbObjects.Add(dbObject);
        }
    });

    await Task.Run(() =>
    {
        // (3) Thread A - a threadpool thread,
        // triggers subscription to DataSource
        // UI Thread is *NOT BLOCKED* due to await
        observableData.Connect()
    });
    // (13) UI Thread - Dispatcher call back here at end of Create call
    // BUT UI THREAD WAS NOT BLOCKED!!!

    // (14) UI Thread - This task will be already completed
    // It is causing a second subscription to the already completed published observable
    await dispatcherObs.ToTask();


}

public class CatalogService
{
   ...

   public IObservable<DbObject> DataSource
   {
        get
        {
            return Observable.Create<DbObject>(obs =>
            {
                // (4) Thread A runs Database query synchronously
                var cursor = Database.Instance.GetAllObjects();
                var status = cursor.MoveToFirst();

                while (status == DbStatus.OK)
                {
                    var dbObject= Db.Create(cursor);
                    // (5) Thread A call OnNext
                    obs.OnNext(dbObject);

                    status = cursor.MoveToNext();
                }

                obs.OnCompleted();
                // (12) Thread A finally completes subscription due to Connect()
                return Disposable.Empty;
            });
        }
    }
}

我认为问题在于大型缓冲区一次性将大量结果卸载到 ObservableCollection 中，为列表框创建了大量工作。