Буфер данных из курсора базы данных при сохранении пользовательского интерфейса

Question

У меня есть каталог базы данных, который заполнен, и курсор, который можно использовать для извлечения объектов. Этот каталог, очевидно, может быть очень большим, и я хотел бы использовать ReactiveUI для буферизации данных, сохраняя при этом привязку данных и чувствительность к пользовательскому интерфейсу. Я последовал шаги 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 события), я подписавшись на источник данных и с использованием метода буфера в надежды на отзывчивость пользовательского интерфейса.

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), он не показывает никаких данных до тех пор, пока весь каталог не будет перечислен. Только тогда обновляется пользовательский интерфейс.

Я новичок в ReactiveUI, но я уверен, что он способен на эту задачу. Есть ли у кого-нибудь предложения или указатели, если я неправильно их использую?

Answer 1

В ожидании дополнительной информации, я думаю, у вас есть несколько буферов нулевой длины в зависимости от того, как долго выполняется запрос БД, за которым следует ровно один ненулевой буфер, содержащий все результаты. Вероятно, вам лучше ограничивать размер буфера как по длине, так и по времени.

EDIT - Я просто хотел провести анализ различных потоков, участвующих в первоначальной реализации. Я не согласен с анализом Павла, я не верю, что поток пользовательского интерфейса заблокирован из-за запроса БД. Я считаю, что он заблокирован из-за буферизации большого количества результатов.

Charlie - пожалуйста, можете ли вы запросить запрос БД в коде (а не отладчиком) и сбросить длины буфера, которые вы получаете тоже.

Я аннотировать код, чтобы показать порядок всех трех потоков, участвующие:

Прежде всего, за пределами предоставленного кода, я предполагаю, что вызов сделан в OnLoad через Loaded события.

(1) - UI Thread вызывает 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 за один раз, создавая тонну работы для ListBox.

Answer 2

Ваша проблема здесь:

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

       status = cursor.MoveToNext(); 
      } 

Этот цикл выполняется синхронно, как только кто-то выписывает, в блокирующем пути. Поскольку вы создаете подписку на поток пользовательского интерфейса (в то время, когда вы вызываете Connect), он будет запускать всю вещь в потоке пользовательского интерфейса. Измените его на:

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; 
});