mongodb - Meteor Mongo Collections find forEach 游标迭代并保存到 ElasticSearch 问题

Question

我有连接到 MongoDB 的 Meteor 应用程序。
在 mongo 中，我有一个包含 ~700k 记录的表。
我每周都有一个 cron 作业，我从表中读取所有记录(使用 Mongo 光标)，并且我想以 10k 的批处理将它们插入到 Elastic Search 中，以便对它们进行索引。

let articles = []
Collections.Articles.find({}).forEach(function(doc) {
        articles.push({ 
            index: {_index: 'main', _type: 'article', _id: doc.id }
        },
        doc);
        if (0 === articles.length % 10000) {
          client.bulk({   maxRetries: 5,  index: 'main',   type: 'article',    body: articles   })
          data = []
        }
    })

由于 for each 是同步的，在继续之前遍历每条记录，并且 client.bulk 是异步的，这会使 Elasticsearch 服务器重载，并且会因内存不足异常而崩溃。
有没有办法在插入完成期间暂停 forEach ？我尝试了异步/等待，但这似乎也不起作用。

let articles = []
Collections.Articles.find({}).forEach(async function(doc) {
        articles.push({ 
            index: {_index: 'main', _type: 'article', _id: doc.id }
        },
        doc);
        if (0 === articles.length % 10000) {
          await client.bulk({   maxRetries: 5,  index: 'main',   type: 'article',    body: articles   })
          data = []
        }
    })

有什么办法可以做到这一点？
编辑:我正在尝试实现这样的目标 - 如果我使用 promise

let articles = []
Collections.Articles.find({}).forEach(function(doc) {
        articles.push({ 
            index: {_index: 'main', _type: 'article', _id: doc.id }
        },
        doc);
        if (0 === articles.length % 10000) {
        // Pause FETCHING rows with forEach
            client.bulk({   maxRetries: 5,  index: 'main',   type: 'article',    body: articles    }).then(() => {
                                          console.log('inserted')
                                          // RESUME FETCHING rows with forEach
                                          console.log("RESUME READING");  
            })
          data = []
        }
    })

Answer 1

设法使它与 ES2018 异步迭代一起工作
有一个想法来自
Using async/await with a forEach loop
这是正在运行的代码

let articles = []
let cursor = Collections.Articles.find({})

for await (doc of cursor) {
    articles.push({ 
        index: {_index: 'main', _type: 'article', _id: doc.id }
    },
    doc);
    if (articles.length === 10000) {
        await client.bulk({   maxRetries: 5,  index: 'trusted',   type: 'artikel',    body: articles   })
        articles = []
    }
}

这工作正常，它设法将所有记录插入 Elastic Search 而不会崩溃。