NodeJS 在多次请求后无限期挂起

Question

我有一个 nodeJS(v.0.10.23) 代理连接到 postgres 数据库（node-postgres 模块 v2.1.0），以及返回各种 json 数据的 pgpool-II。

过去，这是处理连接错误的方式：

var after = function(callback) {
    return function(err, queryResult) {
        if(err) {
            response.writeHead(500, _header);
            console.log("ERROR: 500");
            console.log(err);
            return response.end(JSON.stringify({error: err}));
        }
        callback(queryResult)
    }
};

基本上它的作用是在不存在错误的情况下消耗响应。

可以在这里找到深入的解释：Node js - http.request() problems with connection pooling

使用上面的函数，我得到了这样的结果：

pg.connect(_conString, after(function(err, client, done) {
  client.query(sql, after(function(result) {
  ...
  done();
} 

由于当函数被传递到 after()s 回调时上下文丢失，我失去了使用 pg.connect() 传递的先天 done() 方法的能力。

删除 after 解决了问题，但是，在适当的时候，随着相当多的客户端拉数据，节点将挂起，直到它被重置。

是否有不同的方式来使用各种异步响应？

或者可能是一种将 pg.connect 上下文传递给回调的方法？

Answer 1

见最后一个例子（使用工厂函数和传递响应） 使用工厂函数很有趣，因为它可以节省代码

app.get('/parallel',function(request,response,next){  

function err(err)
{
  console.log(err.stack);
  response.end('error'+err.stack);
}
function send(data)
{
  response.end(JSON.stringify(data))
}

pg.connect(config, function(err, client, done) {
  if(err){done();return err(err);}
  client.query("SELECT * FROM NOW()", function(err, result) {
      done();
      if(err){return err (err);}
      send(result);
    }, 0)
  })
})

您可以尝试使用 colan 的 async to 来选择。

npm 安装异步

// an example using an object instead of an array

var x1=null;
async.series({// in javascript the creation order is preserved in Object.keys
one: function(callback){
    setTimeout(function(){
        x1=3;
        callback(null, 1);
    }, 200);
},
two: function(callback){
    console.log('select where x1='+x1);
    setTimeout(function(){
        callback(null, 2);
    }, 100);
}
},
function(err, results) {// called instantly if there is an error
// results is now equal to: {one: 1, two: 2}
});

单选工厂回调

function makesenderr(response){
 return function senderr(err,ok)
 {
  console.log(err.stack);
  response.end('error'+err.stack);
 }
}

function makesendjson(response){
 return function sendjson(data)
 {
  response.end(JSON.stringify(data))
 }
}


function tryconn(err,ok){
  return function(errarg, client, done) {
   if(errarg){done();return err(errarg);}
   ok(client,done);
  }
}


function doneerrok(done,err,ok){
  return function(errarg, result) {
   done();
   if(errarg){return err(errarg);}
   ok(result);
  }
}

var async=require('async')

app.get('/foo',function(request,response,next){
 var senderr=makesenderr(response)
 var sendjson=makesendjson(response)    
 pg.connect(config, tryconn(senderr,function(client,done){
      client.query("SELECT one FROM t1",doneerrok(done,senderror,sendjson), 0)
 }))
})

使用工厂函数并传递响应

 function senderr(response,err)
 {//add headers here
  console.log(err.stack);
  response.end('error'+err.stack);
 }


 function sendjson(response,data)
 {//add headers here
  response.end(JSON.stringify(data))
 }


function tryconn(response,ok){
  return function(err, client, done) {
   if(err){done();return senderr(response,err);}
   ok(client,done);
  }
}

function donerespok(done,response,ok){
  return function(err, result) {
   done();
   if(err){return err(response,err);}
   ok(response,result);
  }
}

function respok(response,ok){
  return function(err, result) {
   done();
   if(err){return err(response,err);}
   ok(response,result);
  }
}

function donecb(done,cb){ return function(){done();cb.apply(this,arguments);} }

var async=require('async')

app.get('/foo',function(request,response,next){  
 pg.connect(config, tryconn(response,function(client,done){
      client.query("SELECT one FROM t1",donerespok(done,response,sendjson), 0)
 }))
})

app.get('/series',function(request,response,next){  
 pg.connect(config, tryconn(response,function(client,done){

    var one={};
    async.series({ 
        one: function(cb){ 
                  client.query("SELECT one FROM t1", function(err, result) {
                   one=result;cb(err,result);
                  }, 0)
             },
        two: function(cb){
                  client.query("SELECT two FROM t2 where one="+one[0].one,cb, 0)
        }
    },  // results is now equal to: {one: [{one:1}], two: [{two:2},{two:2}]}
    donerespok(done,response,sendjson) );

 }))
})

app.get('/parallel',function(request,response,next){  

    async.parallel({ 
        one: function(cb){ 
         pg.connect(config, tryconn(response,function(client,done){
              client.query("SELECT one FROM t1",donecb(done,cb), 0)
         }))
        },
        two: function(cb){
         pg.connect(config, tryconn(response,function(client,done){
              client.query("SELECT two FROM t2",donecb(done,cb), 0)
         }))
        }
    },  // results is now equal to: {one: [{one:1},{one:2}], two: [{two:1},{two:2}]}
    respok(response,sendjson) );
})

Answer 2

好吧，你当然会失去done()，你永远不会在你的after()函数中将第三个参数传递给你的回调。

function after(cb) {
    return function() {
        // you're using this function in a context where the arguments
        // passed in could be anything. the only constant is that the first
        // argument is the error, if any
        if (arguments[0]) {
            response.writeHead(500, _header);
            console.log("ERROR: 500");
            console.log(err);
            return response.end(JSON.stringify({error: arguments[0]}));
        }
        // apply the entire argument list to the callback, without modification
        cb.apply(cb, arguments);
    };
}

...这也修正了通过queryResult 变量传递client 的可疑约定。

Answer 3

首先，如果您还没有增加连接池，您应该增加连接池，这看起来您不希望限制为 6 个连接。您还应该为请求设置一个相当低的超时时间。

关于数据的上下文，你有没有考虑将pg.connect的this绑定到after函数？这将允许您在本地环境中访问 done。

pg.connect(_conString, after(function(err, client, done) {
  this.done = done;
  client.query(sql, (after(function(result) {
     ...
      this.done();
  }).bind(this));
});

Smashing 杂志前几天有一篇关于使用bind() 的好文章here