Node.js、Mongo async.js 插入和查询

Question

我想要完成的事情： 我查询了一个包含大量文档的集合（对话）。 对于集合中的每个文档/对话，我想查询另一个集合（用户），以查看是否存在与该对话中的 ID 属性匹配的现有用户记录。所以基本上我想看看是否存在附加到对话的用户的用户记录。

Users = { uid:someNumber, 一堆其他属性};

我知道这是 node.js 的异步特性的问题。我一直在尝试使用 async.js 通过回调来解决这个问题。但我想我可能有错，或者没有正确使用它。

问题是对话数组中的每个对话项目都在查询一个项目，但是，由于“保存”尚未完成，“查找”查询永远不会看到有记录已经插入。这是我的代码。也许我在做一些明显错误的事情？所以本质上，检查对话记录，如果用户记录与对话记录上的用户重合，则不做任何事情，如果用户记录不存在，则创建记录。

Conversations.find().limit(1000).exec(function (err, data) {
    //data is an array of conversations, i want to loop through each conversation and compare one of the attribute with an attribute on the Users table
    async.each(data, function(item, callback1){
        //item is a single conversation, on this item there is a participants object that holds two user objects(name, id, type)
        async.each(item.participants, function(user, callback2){

            //this is where i do my query to see if a user exists
            Users.find({uid:user.participantId}).exec(function (err, results){
                //if the user doesn't exist then create a user record
                if(results.length == 0){
                    var user = new Users();
                    user.name =user.participantName;
                    user.uid = user.participantId;
                    user.type = user.participantType;

                    user.save(function(err, result){
                        console.log(result);
                        //after it has saved, callback2() so that the second item in the array will query against the Users table
                        callback2();
                    })
                }
                else{
                  callback2()
            })

        })
        //first item in the conversations array is completed, callback1(), second item should now start
        callback1();

    });
})

Answer 1

您可以通过实现"stream" 处理以及使用.findOneAndUpdate() 来大量清理并节省内存使用：

var stream = Conversations.find().stream();

stream.on("data",function(item) {
    stream.pause();                 // pauses processing stream

    async.each(
        item.particpants,
        function(user,callback) {
            Users.findOneAndUpdate(
                { "uid": user.participantId },
                { "$setOnInsert": {
                    "name": user.participantName,
                    "type": user.participantType
                }},
                { "upsert": true, "new": true },
                callback
            );
        },
        function(err) {
            if (err) throw err;
            stream.resume();        // resume stream
        }
    );

});

stream.on("error",function(err) {
    // error handling
});

stream.on("end",function() {
    // Complete
});

基本上，您可以通过实现流来避免将所有结果从Conversations 加载到内存中（猫鼬默认）。然后在从流结果中读取每个项目时，您会处理.findOneAndUpdate()，它会查找存在的项目并返回修改后的结果。

{ "upsert": true } 表示如果找不到它，则会在集合中创建一个新文档。 $setOnInsert 是一个 MongoDB 修饰符，它确保“更新”更改仅在创建新文档时应用，因此它不会在找到现有文档时更改。

当然，这也可以是.update()，您对结果不做任何事情（因为这实际上并没有对结果做任何事情），但我要离开.findOneAndUpdate()，以防万一您想console.log() 看看发生了什么。使用.update()会更高效，因为不需要返回文档，并且基本上采用相同的参数。

除了内部的async.each 流控制之外，还有.pause() 和.resume() 的流控制。这些本质上控制外部条目的流程，一次允许一个项目。您可以对此进行扩展以允许并行处理一组项目，但这是基本示例。

当然，一个事件流也会告诉你它何时完成，并且由于另一个流控制已经在处理其他异步操作，所以只有在所有项目都完成时才会调用它。