节点中的并发请求处理。js和MongoDB

I have a Node.js server which has to handle 6 concurrent requests. Each request reads a document from MonogDB database and then modifies it.

我有一个节点。js服务器，需要处理6个并发请求。每个请求从一夫一妻数据库中读取一个文档，然后修改它。

So, when we send 6 requests concurrently they all read the same data item and modify it but I want that value modified by the first request should be read by the second request and then it should be able to modify it further and so on.

因此，当我们同时发送6个请求时，它们都读取相同的数据项并进行修改，但是我希望第一个请求修改的值应该被第二个请求读取，然后它应该能够进一步修改它，以此类推。

For example, I have a data item called, x with initial value 'a'. The first request modifies it to 'ab'. So, the second request now should read 'ab' and not 'a'. But it always reads 'a' because they are coming concurrently.

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

I have tried to apply semaphore, mutex, async.series, async.waterfall, sleep and promises but nothing worked. Probably I am not using them correctly. So, how should they be used to handle these concurrent requests in a synchronous manner.

我尝试过应用信号量、互斥量和异步。系列,异步。瀑布，睡眠和承诺，但什么都没用。可能我没有正确地使用它们。因此，应该如何使用它们以同步的方式处理这些并发请求。

Or is there any other way of doing it? Can we do something with timestamp ordering protocol? If yes, then how can I apply it in Node.js?

或者还有别的方法吗?我们可以用时间戳订购协议吗?如果是，我如何在Node.js中应用它?

This is the code that I want to run for these concurrent requests-

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

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");
                    }
                ); 
            });
    });
});

Here I am sending requests like - 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, etc. concurrently.

我正在发送这样的请求: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,等等。

Now, they can run in any order. But I want all 'b', 'c', 'd' etc to be appended. And thus the output for each should look somewhat like - 'ab', 'abc', 'abcd' and so on.

现在，它们可以以任何顺序运行。但是我想要所有的b, c, d等等都加在后面。因此，每个函数的输出应该看起来像- ab, abc, abcd等等。

But the output I get is- 'ab', 'ac', 'ad'.

但是输出是- ab, ac, ad。

How exactly should I use promise to make sure these requests execute in a synchronous manner?

我应该如何准确地使用promise来确保这些请求以同步的方式执行?

3 个解决方案

#1

It sounds like a complex system. My suggestion would be to take a look at the new Mongo Node driver. Then take a look at the ECMAScript6 documentation. This involves installing the co module it works with. Use co as you would use promises (take a look at the module). It supplies you with yield which returns a promise.

这听起来像是一个复杂的系统。我的建议是查看新的Mongo节点驱动程序。然后查看ECMAScript6文档。这涉及到安装它所使用的co模块。像使用承诺一样使用co(查看模块)。它提供给你回报承诺的收益。

It also sounds like you will need to aggregate (if I am not mistaken.) For a then ab then abc and so on.

听起来你也需要集思广益(如果我没弄错的话)。a然后ab然后abc等等。

The concurrency problem should be handled obviously by waiting for the yield to return. The examples in the new documentation are not great but decent enough for you to get a working example out of them.

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

#2

You need to implement a semaphore.

您需要实现一个信号量。

Read about it here

读到这里

In your case you have to serve one request at a time though they come in "all together". So

在你的情况下，你必须一次服务一个请求，尽管他们“一起”来。所以

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);
      });
    });
});

#3

So, all I had to do was to implement semaphore correctly. This is the final version of the code that worked-

我所要做的就是正确地实现信号量。这是工作代码的最终版本。

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();    

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

#1