Node.js async.series() 不要等待回调完成

Question

当mongoose.disconnect() 从未被调用并且节点程序有时间执行所有回调时，我的代码可以工作。当在程序结束时调用mongoose.disconnect() 并且连接丢失时，这些对 mongo 的插入自然不会发生。

此异步不等待这些插入完成。为什么，我该如何解决？

mObjectList = 准备插入的对象数组

insertMongoObj: function(mObjectList, callback) {
  var tasks = [];

  for (i in mObjectList) {
    tasks.push(mObjectList[i].save());
  }

  async.parallel(tasks, function(err) {
    if (err) {
      callback(err, 0);
    }
  });
  callback(null, tasks.length);
}

感谢您的快速回复！按照建议更改了功能，但同样的事情。当连接永远不会关闭并且程序永远不会退出时工作。 Mongoose.disconnect() 在这些插入之前仍在关闭连接。

insertMongoObj : function(mObjectList, callback){

    var tasks = [];

    for (i in mObjectList) {
      tasks.push(mObjectList[i].save.bind(mObjectList[i]));
    }

    async.parallel(tasks, function(err) {
        if(err) return callback(err);
        winston.info("  file processing ready");
    });

    return callback();
}

---

之前的函数调用在这里：

mongoose.connection.once('open', function () {
    talendToMongo.processTalendExport(objPaths, function(){
        mongoose.disconnect();
    });

});

processTalendExport : function(talendEntityLocationList, callback){
    var mongoObjectList = [];
    var self = this;
    for(obj in talendEntityLocationList){
        winston.info("**Start processing directory:" + talendEntityLocationList[obj]);
        var fileList = fs.readdirSync(talendEntityLocationList[obj]);
        this.processEntityDir(talendEntityLocationList[obj], fileList, function(mongoObjectList){
            self.insertMongoObj(mongoObjectList, function(err){
                if(err) {
                    winston.error("  Error processing file:" + talendEntityLocationList[obj]);
                        process.exit(0);
                    } else {
                        winston.info(" Mongo Objects inserted");
                    }
                });
            });
        }
        callback();
    }

insertMongoObj : function(mObjectList, callback){
    var tasks = [];

    for (i in mObjectList) {
        tasks.push(mObjectList[i].save.bind(mObjectList[i]));
    }

    async.parallel(tasks, function(err) {
        if(err) return callback(err);
          winston.info("  file processing ready");
        });

        return callback();
    }

Answer 1

您需要将最后一次回调调用移动到 async.parallel 的回调中：

for (i in mObjectList) {
  tasks.push(mObjectList[i].save.bind(mObjectList[i]));
}

async.parallel(tasks, function(err) {
     if (err) {
         return callback(err, 0);
     }
     return callback(null, tasks.length);
 });

Answer 2

这是一个很好的例子，async.each 是理想的。

insertMongoObj: function(mObjectList, callback) {
    async.each(mObjectList, function(obj, done){
        obj.save(done);
    }, callback);
}

更新

在processTalendExport 中，callback() 在for(obj in talendEntityLocationList){ 循环之后而不是在插入完成之后，因此它会在插入完成之前调用.disconnect。取决于客户端和网络，这可能总是会导致失败，或者只是有时会失败。您也应该使用async.each(talendEntityLocationList, ... 来查看该列表。

Answer 3

终于让这个工作更正了以前的函数回调并修复了 async.eachSeries 回调：

    insertMongoObjects : function(mongoObjList, callback){
        var insertCnt = 0;
        winston.info("    Start mongo object insert");
        async.eachSeries(mongoObjList,
            function(mongoObj, done){
                    mongoObj.save(function(save_result){
                        if(!save_result){ 
                            insertCnt++
                            done(null);
                        } else { done(save_result); }
                    });
            },
            function(err){
                if(err) {
                    winston.error("    Error inserting mongo object:");
                    winston.error(err);
                    callback(err);
                } else {
                    winston.info("    Mongo object insert successful for:" + insertCnt);
                    callback();
                }
            }
        );
    }