node.js - 以 react 方式查询 mongodb 集合中所有文档的最佳方法，无需溢出 RAM

Question

我想以 react 方式查询集合中的所有文档。 collection.find() mongodb nodejs 驱动程序的方法返回一个游标，该游标为集合中找到的每个文档触发事件。所以我做了这个:

function giant_query = (db) => {
    var req = db.collection('mycollection').find({});   
    return Rx.Observable.merge(Rx.Observable.fromEvent(req, 'data'),
                               Rx.Observable.fromEvent(req, 'end'),
                               Rx.Observable.fromEvent(req, 'close'),
                               Rx.Observable.fromEvent(req, 'readable'));
}

它会做我想做的事:为每个文档触发，这样我就可以以一种 react 性的方式处理，就像这样:

Rx.Observable.of('').flatMap(giant_query).do(some_function).subscribe()

我可以以数十个数据包的形式查询文档，但是每次触发可观察流时我都必须跟踪索引号，并且我必须制作一个我不知道的可观察循环如果可能或正确的方法。

这个游标的问题是我不认为它在数据包中做事。它可能会在短时间内触发所有事件，因此会淹没我的 RAM。即使我使用 Observable 的缓冲区在数据包中缓冲一些事件，事件和事件数据(文档)也将在 RAM 上等待被操作。

以被动方式处理它的最佳方式是什么？

Answer 1

我不是 mongodb 方面的专家，但根据我见过的例子，这是我会尝试的模式。

我省略了数据以外的事件，因为节流事件似乎是主要问题。

var cursor = db.collection('mycollection').find({});  

const cursorNext = new Rx.BehaviourSubject('next');  // signal first batch then wait
const nextBatch = () => {
  if(cursor.hasNext()) {
    cursorNext.next('next');
  }
});

cursorNext
  .switchMap(() =>                            // wait for cursorNext to signal
     Rx.Observable.fromPromise(cursor.next())  // get a single doc
       .repeat()                               // get another
       .takeWhile(() => cursor.hasNext() )     // stop taking if out of data
       .take(batchSize)                        // until full batch
       .toArray()                              // combine into a single emit
  )
  .map(docsBatch => {
    // do something with the batch
    // return docsBatch or modified doscBatch
  })
  ... // other operators?
  .subscribe(x => {
    ...
    nextBatch();
  });         

我正在尝试在没有 mongodb 的情况下对这个 Rx 流程进行测试，同时这可能会给你一些想法。