2

私はもうしばらくの間、.Net で RabbitMQ を使用してきましたが、あまり問題はありません。現在、node.js を使用して rabbit.js に移行していますが、あまり詳しくありません。rabbit.js のドキュメントは限られています。私が知っているのは、基本的な PUSH/PULL または PUB/SUB だけです。REQ/REP を実行したいのですが、方法がわかりません。誰でもスニペットを共有できます。

ご返信いただきありがとうございます。

一番、

4

2 に答える 2

14

これはおそらくあなたが要求した以上のものですが、rabbit.jsで REQ/RES の代わりにnode-amqpを使用して RPC を実行するためのスニペットがあります (かなり長いですが)。私が行ったことは、RPC に関するRabbitMQ チュートリアルで見つけることができるものと似ています

今のところ、メッセージの内容は、amqp モジュールによって json に変換されるオブジェクト (ハッシュ) である必要があります。

AmqpRpc クラスは、初期化時に amqp 接続を取得するため、makeRequest を呼び出してコールバックで応答を待つだけで済みます。応答の形式はfunction(err, response)で、err はタイムアウト エラーの可能性があります

申し訳ありませんが、あなたが求めていたものとは正確には一致しませんが、おそらく十分に近いものです。また、コードをGistとしてgithubに投稿しました：https://gist.github.com/2720846

編集: 複数の未処理のリクエストをサポートするようにサンプルが変更されました。

amqprpc.js

var amqp = require('amqp')
  , crypto = require('crypto')

var TIMEOUT=2000; //time to wait for response in ms
var CONTENT_TYPE='application/json';

exports = module.exports = AmqpRpc;

function AmqpRpc(connection){
  var self = this;
  this.connection = typeof(connection) != 'undefined' ? connection : amqp.createConnection();
  this.requests = {}; //hash to store request in wait for response
  this.response_queue = false; //plaseholder for the future queue
}

AmqpRpc.prototype.makeRequest = function(queue_name, content, callback){
  var self = this;
  //generate a unique correlation id for this call
  var correlationId = crypto.randomBytes(16).toString('hex');

  //create a timeout for what should happen if we don't get a response
  var tId = setTimeout(function(corr_id){
    //if this ever gets called we didn't get a response in a 
    //timely fashion
    callback(new Error("timeout " + corr_id));
    //delete the entry from hash
    delete self.requests[corr_id];
  }, TIMEOUT, correlationId);

  //create a request entry to store in a hash
  var entry = {
    callback:callback,
    timeout: tId //the id for the timeout so we can clear it
  };

  //put the entry in the hash so we can match the response later
  self.requests[correlationId]=entry;

  //make sure we have a response queue
  self.setupResponseQueue(function(){
    //put the request on a queue
    self.connection.publish(queue_name, content, {
      correlationId:correlationId,
      contentType:CONTENT_TYPE,
      replyTo:self.response_queue});
  });
}


AmqpRpc.prototype.setupResponseQueue = function(next){
  //don't mess around if we have a queue
  if(this.response_queue) return next();

  var self = this;
  //create the queue
  self.connection.queue('', {exclusive:true}, function(q){  
    //store the name
    self.response_queue = q.name;
    //subscribe to messages
    q.subscribe(function(message, headers, deliveryInfo, m){
      //get the correlationId
      var correlationId = m.correlationId;
      //is it a response to a pending request
      if(correlationId in self.requests){
        //retreive the request entry
        var entry = self.requests[correlationId];
        //make sure we don't timeout by clearing it
        clearTimeout(entry.timeout);
        //delete the entry from hash
        delete self.requests[correlationId];
        //callback, no err
        entry.callback(null, message);
      }
    });
    return next();    
  });
}

それを使用する方法の小さな例を以下に示します。両方のコード部分を保存し、実行するだけです...

node client.js

応答を提供するサーバーがない場合、要求はタイムアウトします。

client.js

//exmaple on how to use amqprpc
var amqp = require('amqp');
var connection = amqp.createConnection({host:'127.0.0.1'});

var rpc = new (require('./amqprpc'))(connection);

connection.on("ready", function(){
  console.log("ready");
  var outstanding=0; //counter of outstanding requests

  //do a number of requests
  for(var i=1; i<=10 ;i+=1){
    //we are about to make a request, increase counter
    outstanding += 1;
    rpc.makeRequest('msg_queue', {foo:'bar', index:outstanding}, function response(err, response){
      if(err)
        console.error(err);
      else
        console.log("response", response);
      //reduce for each timeout or response
      outstanding-=1;
      isAllDone();
    });
  }

  function isAllDone() {
    //if no more outstanding then close connection
    if(outstanding === 0){
      connection.end();
    }
  }

});

念のため、サンプルサーバーも投入します

サーバー.js 

//super simple rpc server example
var amqp = require('amqp')
  , util = require('util');

var cnn = amqp.createConnection({host:'127.0.0.1'});

cnn.on('ready', function(){
  console.log("listening on msg_queue");
  cnn.queue('msg_queue', function(q){
      q.subscribe(function(message, headers, deliveryInfo, m){
        util.log(util.format( deliveryInfo.routingKey, message));
        //return index sent
        cnn.publish(m.replyTo, {response:"OK", index:message.index}, {
            contentType:'application/json',
            contentEncoding:'utf-8',
            correlationId:m.correlationId
          });
      });
  });
});
于 2012-05-17T19:19:34.937 に答える
2

https://github.com/demchenkoe/node-amqp-rpc

これは、ノードで rabbitMQ を介して rpc を実行するのに最適であることがわかりました。強くお勧めします。

于 2012-11-27T05:30:07.860 に答える