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

Question

我有一个已填充的数据库目录和一个可用于检索对象的游标。这个目录显然可能非常大，我想做的是使用 ReactiveUI 来缓冲数据，同时保持 UI 数据绑定和响应式。我按照here 的步骤将我的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 的新手，但我确信它能够胜任手头的任务。如果我使用不正确，有人有什么建议或指点吗？

Answer 1

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

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

Charlie - 请您在代码中（而不是使用调试器）对 DB 查询进行计时并转储您获得的缓冲区长度。

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

首先，在提供的代码之外，我假设通过Loaded 事件调用OnLoad。

(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 中，从而为列表框创建大量工作。

Answer 2

你的问题在这里：

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

                status = cursor.MoveToNext();
            }

一旦有人订阅，该循环就会以阻塞方式同步运行。由于您在 UI 线程上创建订阅（在您调用 Connect 时），它将在 UI 线程上运行整个事情。将其更改为：

return Observable.Create<DbObject>(obs =>
{
    Observable.Start(() => {
        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();
    }, RxApp.TaskPoolScheduler);

    return Disposable.Empty;
});