RabbitMQ请求，总是超时

Question

我有一个奇怪的问题，我的回调从未发布，消息超时，即使方法在队列中运行。这发生在一些特定的队列中，在它发生一次之后，我不能从客户端发出任何其他请求，即使以前工作过，它们都超时了。必须重新启动客户端和服务器才能使其再次工作。这就是代码，它发生的地方，我似乎不明白哪里出错了。

Server.js文件，我正在创建队列。我有几个这样的队列，这是其中之一。

var amqp = require('amqp');
var util = require('util');
var cnn = amqp.createConnection({host:'127.0.0.1'});
var getCart = require('./services/getCart');
cnn.on('ready', function() {
    cnn.queue('getCart_queue', function(q){
            q.subscribe(function(message, headers, deliveryInfo, m){
                // util.log(util.format( deliveryInfo.routingKey, message));
                // util.log("Message: "+JSON.stringify(message));
                // util.log("DeliveryInfo: "+JSON.stringify(deliveryInfo));
                getCart.handle_request(message, function(err,res){
                    cnn.publish(m.replyTo, res, {
                        contentType:'application/json',
                        contentEncoding:'utf-8',
                        correlationId:m.correlationId
                    });
                });
            });
        });
});

这里，句柄请求函数成功完成，但回调函数从未通过，并且总是在另一端超时

var cart = require('../models/cart');
function handle_request(msg, callback) {
    var user_id = msg.id;
    cart
        .find({id:user_id})
        .populate('users ads')
        .exec(function(err, results){
                       // This works, just the callback doesnt
            if(!err){
                console.log(results);
                callback(null, results);
            } else {
                console.log(err);
                callback(err, null);
            }
        });
}
exports.handle_request = handle_request;

这是我调用请求的方式

 var msg_payload = {"id":id};
    mq_client.make_request('getCart_queue', msg_payload, function(err, results){
      console.log(results); // never prints 
       //stuff that is never reached
});

这些是我的rpc文件，我不认为这些应该有任何问题，因为其他一些队列工作正常。这是client

显示的错误

GET /getCart - - ms - -
Error: timeout 6ee0bd2a4b2ba1d8286e068b0f674d8f
    at Timeout.<anonymous> (E:Ebay_clientrpcamqprpc.js:32:18)
    at Timeout.ontimeout [as _onTimeout] (timers.js:341:34)
    at tryOnTimeout (timers.js:232:11)
    at Timer.listOnTimeout (timers.js:202:5)

希望信息不是模糊的，如果你需要更多的信息，请告诉我。谢谢!

我认为错误是在这个文件中，因为我尝试调试和从rabbitmq服务器，回调正在被调用，它有相关id以及对变量的回复，所以请求没有在这里被拾取。

var amqp = require('amqp')
  , crypto = require('crypto');
var TIMEOUT=8000; 
var CONTENT_TYPE='application/json';
var CONTENT_ENCODING='utf-8';
var self;
exports = module.exports = AmqpRpc;
function AmqpRpc(connection){
  self = this;
  this.connection = connection; 
  this.requests = {}; 
  this.response_queue = false; 
}

AmqpRpc.prototype.makeRequest = function(queue_name, content, callback){
  self = this;
  var correlationId = crypto.randomBytes(16).toString('hex');

  var tId = setTimeout(function(corr_id){
    callback(new Error("timeout " + corr_id));
    delete self.requests[corr_id];
  }, TIMEOUT, correlationId);
  var entry = {
    callback:callback,
    timeout: tId 
  };
  self.requests[correlationId]=entry;

    self.setupResponseQueue(function(){

    self.connection.publish(queue_name, content, {
      correlationId:correlationId,
      contentType:CONTENT_TYPE,
      contentEncoding:CONTENT_ENCODING,
      replyTo:self.response_queue});
  });
};

AmqpRpc.prototype.setupResponseQueue = function(next){
  if(this.response_queue) return next();
  self = this;
  self.connection.queue('', {exclusive:true}, function(q){  
    self.response_queue = q.name;
    q.subscribe(function(message, headers, deliveryInfo, m){
      var correlationId = m.correlationId;
      if(correlationId in self.requests){
        var entry = self.requests[correlationId];
        clearTimeout(entry.timeout);
        delete self.requests[correlationId];
        entry.callback(null, message);
      }
    });
    return next();    
  });
};

Answer 1

这是make_request()在client.js文件中的代码:

var amqp = require('amqp');
var connection = amqp.createConnection({host:'127.0.0.1'});
var rpc = new (require('./amqprpc'))(connection);

function make_request(queue_name, msg_payload, callback){
    rpc.makeRequest(queue_name, msg_payload, function(err, response){
        if(err)
            console.error(err);
        else{
            console.log("response", response);
            callback(null, response);
        }
    });
}
exports.make_request = make_request;

看看rpc.makeRequest()出错时会发生什么:

rpc.makeRequest(queue_name, msg_payload, function(err, response){
    if(err)
        console.error(err);
        //
        //HERE: should be a callback call here. 
        //
    else{
        console.log("response", response);
        callback(null, response);
    }
});

这可能是您获得超时的原因。

Answer 2

rabbitMQ没有问题，但在处理请求和响应请求后，我的查询。

对于遇到此问题的其他人，检查并仔细检查每个语句，因为错误不会显示在控制台中，而只会显示超时