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

Question

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

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

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

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

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 的所有结果加载到内存中(mongoose 默认值)。然后，当从流结果中读取每个项目时，您处理 .findOneAndUpdate()，它会查找存在的项目并返回修改后的结果。

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

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

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

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