Nodejs Async upsert 问题答案

【问题标题】：Nodejs Async upsert issueNodejs Async upsert 问题
【发布时间】：2014-08-15 04:57:00
【问题描述】：

我目前正在做一个使用 MongoDB 的 Nodejs 项目，并密切监视 Nodejs 服务器的数据流。

我的节点服务器的代码如下：

Receive.js
1.节点服务器接收到一个JSON文本文件
2. 节点服务器告诉 MongoDB (upsert) 我们收到了一个文件

Process.js
3. 节点服务器将 JSON 文件更新到 MongoDB
4. 节点服务器告诉 MongoDB 我们处理了所述文件。

问题是有时 #4 出现在 #2 之前，尽管 #1 总是出现在 #3 之前.我的监视器程序开始显示处理的文件多于接收的文件。有没有办法在不使该服务器完全同步的情况下解决此问题？

例如，如果我向我的 Node 服务器发送 500 个 JSON 文本文件，那么 Node 最后一次执行 #2 应该在最后一次执行 #4 之前。

** 注意：App.js 从 Receive.js 调用 .receive 和 Receive.js 从 Process.js 调用 .save **

app.js

http.createServer(app).listen(app.get('port'), function(){
  mongodb.createConnection(function(db){
    if(db){
      console.log('success connecting to mongodb!');
    }
  });
});  
app.post('/log', logCollection.receive);

--
Receive.js

exports.receive = function(req, res, next){
  var usagelog = req.body.log;

  if(!usagelog){
    log4js.logger.error("Usagelog is Null!");
    res.end("fail");
  }else{      
    count++;

    //Upsert the Receive Count for Monitor  
      mongodb.getConnection(function(db){

          dateFormat.initDate();
          //Upsert trip (usage logs)
          var currentDateTime =  moment().format('YYYY-MM-DD hh:mm:ss');
          db.collection("ReceiveCount").update( {"date":dateFormat.currentDate(), "pid":process.pid }, 
                    {"date":dateFormat.currentDate(), "IPAddress": ip.address(), "pid":process.pid, "countReceive" : count, "LastReceivedTime": currentDateTime}, 
                    {upsert:true}, function(err, result) {            
            });
          });

    //End Receive count
    var usagelogJSON = convertToJson(usagelog);
    var usagelogWithCurrentDate = addUpdateTime(usagelogJSON);
    usagelogDao.save(usagelogWithCurrentDate, res);
  }
};

Process.js

exports.save = function(usagelog, res){
  var selector = upsertCondition(usagelog);

  mongodb.getConnection(function(db){
    //Upsert trip (usage logs)
    db.collection(collectionName()).update(selector, usagelog, {upsert:true, fullResult:true}, function(err, result) {
      if(err){
        log4js.logger.error(err);
        res.end("fail");
      }else{

        if(result.nModified == 0)
            countInsert++;
        else
            countUpdate++;

        //Upsert Processed Count (both Updated and Inserted     
        db.collection("ReceiveCount").find({"date":dateFormat.currentDate(), "pid":process.pid },{}).toArray(function (err, docs) {

            var receivecount = docs[0].countReceive;
            var receivetime = docs[0].LastReceivedTime;
            var currentDateTime = moment().format('YYYY-MM-DD hh:mm:ss');
            var MongoCount=0;

            db.collection("raw_"+dateFormat.currentDate()).count(function(err, count){
                console.log("raw_"+dateFormat.currentDate());

                MongoCount = count;
                console.log("Mongo count is :" +MongoCount);
            });

            db.collection("ReceiveCount").update( {"date":dateFormat.currentDate(), "pid":process.pid }, 
                {"date":dateFormat.currentDate(), "IPAddress": ip.address(), "pid":process.pid, "countUpdate" : countUpdate, "countInsert":countInsert, "countTotal":countUpdate+countInsert, "LastProcessTime": currentDateTime,
                "countReceive":receivecount, "LastReceivedTime":receivetime, "MongoCount":MongoCount}, {upsert:true}, function(err, result) {      
            }); 
        });
        res.end("success");
      }
    });
  });
};

【问题讨论】：

尽管这里有赞成票，但用 SO 术语来说并不是一个非常明确的问题。有一些方法可以通过异步处理和库来处理这个问题。但是我们需要看到一些代码才能成为一个可以回答和解决的问题。显示您的代码，以便我们可以看到您哪里出错了。点赞！= 答案。
更适合更新，但我认为您如何调用“接收”和“保存”导出的上下文是您问题中更重要的部分。

标签： javascript jquery ajax node.js mongodb

【解决方案1】：

您可以使用异步模块 (https://github.com/caolan/async)。这是伪代码：

async.parallel([
  fun1(cb) {
    tellDBYouReceivedFile(..., onFinished() {
      cb();
    });
  },
  fun2(cb) {
    saveYourFileToDB(..., onFinished() {
      cb();
    });
  },
], function() {
  tellDBYouProcessedFile();
});

您要实现的目标无法完全异步完成，因为最后一步（#4）必须先等待 #2 和 #3 完成。上面的例子可以让#2和#3异步，保证#4最后执行。

【讨论】：