Node.js 和 MongoDB 中的并发请求处理

Question

我有一个 Node.js 服务器，它必须处理 6 个并发请求。 每个请求都会从 MonogDB 数据库中读取一个文档，然后对其进行修改。

所以，当我们同时发送 6 个请求时，它们都读取相同的数据项并对其进行修改，但我希望第一个请求修改的值应该由第二个请求读取，然后它应该能够进一步修改它等等开。

例如， 我有一个名为 x 的数据项，初始值为“a”。 第一个请求将其修改为“ab”。所以，第二个请求现在应该是“ab”而不是“a”。但它总是读取“a”，因为它们是同时出现的。

我尝试应用信号量、互斥锁、async.series、async.waterfall、sleep 和 promise，但没有任何效果。可能我没有正确使用它们。那么，应该如何使用它们来同步处理这些并发请求呢？

或者还有其他方法吗？我们可以用时间戳排序协议做点什么吗？如果是，那么我如何在 Node.js 中应用它？

这是我要为这些并发请求运行的代码-

app.get('/urltest', function (req, res) {
    var q=require('url').parse(req.url,true).query;
    var s=q.s;

    MongoClient.connect(url, function(err, db) {
        if(err)
            throw err;
         db.collection( 'Comp' ).find({no:2}).snapshot().forEach(
            function (elem) {

                db.collection( 'Comp' ).findAndModify (
                    { no : elem.no },
                    [['no', 1]],
                    { $set : { age:elem.age+s } }, {new:true},
                    function( err, result ) {
                        if ( err ) console.log( err);

                        console.log("after update: "+result.value.age);
                        res.end("success");
                    }
                ); 
            });
    });
});

我在这里同时发送诸如 http://192.168.197.140:8081/urltest?s=b、http://192.168.197.140:8081/urltest?s=c、http://192.168.197.140:8081/urltest?s=d 等请求。

现在，它们可以按任何顺序运行。但我希望附加所有“b”、“c”、“d”等。因此每个的输出应该看起来有点像 - 'ab'、'abc'、'abcd' 等等。

但我得到的输出是-'ab'、'ac'、'ad'。

我应该如何使用 Promise 来确保这些请求以同步方式执行？

Answer 1

这听起来像是一个复杂的系统。我的建议是看看新的Mongo Node driver。 然后看看ECMAScript6 documentation。这涉及安装它使用的co 模块。像使用 Promise 一样使用 co（查看模块）。它为您提供返回承诺的 yield。

听起来你还需要聚合（如果我没记错的话）。对于 a 然后 ab 然后 abc 等等。

并发问题显然应该通过等待yield返回来处理。新文档中的示例不是很好，但足以让您从中获得一个工作示例。

Answer 2

所以，我所要做的就是正确实现信号量。 这是代码的最终版本-

var sem = require('semaphore')(1);      
app.get('/urltest', function (req, res) {

    sem.take(function() {

        var q=require('url').parse(req.url,true).query;
        var s=q.s;                                  

        MongoClient.connect(url, function(err, db) {
            if(err)
                throw err;
            db.collection( 'Comp' ).find({no:2}).snapshot().forEach(
                function (elem) {                           

                    db.collection( 'Comp' ).findAndModify (
                        { no : elem.no },
                        [['no', 1]],
                        { $set : { x:elem.x+s } }, {new:true},          
                        function( err, result ) {
                            if ( err ) console.log( err);

                            console.log("after update: "+result.value.x);
                            res.end("success");

                            sem.leave();    

                        }
                    ); 
                });
        });
    })
});

Answer 3

你需要实现一个semaphore。

阅读它here

在您的情况下，您必须一次处理一个请求，尽管它们“一起”出现。所以

var sem = require('semaphore')(1);
var server = require('http').createServer(req, res) {
    sem.take(function() {
      // find, update and commit data before leaving the semaphore
      expensive_database_operation(function(err, res) {
        sem.leave();

        if (err) return res.end("Error");

        return res.end(res);
      });
    });
});