node-amqpとamqplibの書き方の違い

nodeでnode-amqpとamqplibを使った場合のそれぞれの書き方の違い

RabbitMQをローカルホストにはインストール起動済み
npm install amqp
npm install amqplibも終わっているものとしています。

３秒ごとにbrokerに現在時刻を送信する

node-amqp

接続用のオプションはオプションオブジェクトで指定します
送信にはpublish関数を使います

pub1.js

require('date-utils');
var amqp=require('amqp');
var connection = amqp.createConnection({
    host: 'localhost',
    login: 'genpub',
    password: 'rabbitmq'
});
var updater = setInterval(function() {
    var cur= new Date();
    var ts=cur.toFormat("HH24:MI:SS");
    console.log(ts);
    connection.publish('work1',ts,{deliveryMode: 2});
}, 3000);

amqplib

接続用オプションはURLに埋め込みます
channel生成を明示的に行って、チャンネルオブジェクトのsendToQueue関数を使います

pub2.js

require('date-utils');
var amqp = require('amqplib/callback_api');
var ch; 
amqp.connect('amqp://genpub:rabbitmq@localhost', function(err, conn) {
  conn.createChannel(function(err, channel) {
    ch=channel;
    console.log("connected");
  }); 
});
var updater = setInterval(function() {
    var q = 'work1';
    var cur= new Date();
    var ts=cur.toFormat("HH24:MI:SS");
    ch.assertQueue(q, {durable: true});
    ch.sendToQueue(q, new Buffer(ts), {persistent: true});
    console.log(ts);
}, 3000);

ワークキューからメッセージを受信する

ワークキューは、複数の受信者がメッセージを待ち受けている時、誰か一人の受信者だけがメッセージを受け取れます

node-amqp

単純にメッセージを受け取ったらそれを表示して再度待ち受け状態に戻ります。

sub1.js

var amqp = require('amqp');
var connection = amqp.createConnection({
  host: 'localhost',
  login: 'gensub',
  password: 'rabbitmq'
});
connection.on('ready', function() {
  connection.queue('work1', {autoDelete: false, durable: true},
    function(q) {
      q.subscribe(function(message){
        var msg=message.data.toString('utf8');
        console.log(msg);
      }); 
  }); 
});

amqplib

メッセージを受け取ったら、何か重い仕事をする想定として１０秒間スリープします。１０秒後に仕事が終わったものとして、再びメッセージ待ち受け状態に戻ります。

sub2.js

var amqp = require('amqplib/callback_api');
amqp.connect('amqp://gensub:rabbitmq@localhost:5672', function(err, conn) {
  conn.createChannel(function(err, ch) {
    var q = 'work1';
    ch.assertQueue(q, {durable: true});
    ch.prefetch(1);//1メッセージ受け取ったら受信の一時停止を通知 
    ch.consume(q, function(msg) {
      var secs = 10; 
      console.log("Received %s", msg.content.toString());
      setTimeout(function() {
        console.log("Done");
        ch.ack(msg);//再び受信を再開
      }, secs * 1000);
    }, {noAck: false});
  }); 
});

実証実験

第１画面でpub1.jsを起動して３秒ごとに現在時刻をpublishする
第２画面でsub1.jsを起動すると、時刻メッセージをそのまま受信する
第３画面で16:00:20にsub2.jsを起動すると、sub2.jsに16:00:21のメッセージが渡されて、sub1.jsには渡らない。sub2.jsは10秒間メッセージを受け取れないため、その間はsub1が受信する。sub2は10秒後にDoneというメッセージを出して、再びメッセージ受信可能状態に戻る。