Перевод документации модуля

new_gist_file

node-amqp

Это клиент для RabbitMQ (и, возможно, для других серверов?). Этот клиент частично реализует версию 0.9.1 протокола AMQP.

Установка
npm install amqp

Синопсис

Пример подключения к серверу и прослушивание очереди.

var amqp = require('amqp');

var connection = amqp.createConnection({ host: 'dev.rabbitmq.com' });

// Ожидаем соединения
connection.on('ready', function () {
  // Используем 'amq.topic' для обмена по умолчанию
  connection.queue('my-queue', function(q){
      // Перехватываем все сообщения
      q.bind('#');

      // Подписываемся на получение сообщений
      q.subscribe(function (message) {
        // Выводим сообщения
        console.log(message);
      });
  });
});

Подключение

new amqp.Connection() экземпляр нового соединения. Используем connection.connect() для подключения к серверу. amqp.createConnection() возвращает экземпляр amqp.Connection который являеться подклассом net.Stream. Все события и методы которые использует net.Stream могут быть использованы и с amqp.Connection.

Варианты подключение

amqp.createConnection([options, [implOptions]]) принимает два объекта с параметрами. Параметры  по умолчанию: 

{ host: 'localhost'
, port: 5672
, login: 'guest'
, password: 'guest'
, authMechanism: 'AMQPLAIN'
, vhost: '/'
, ssl: { enabled : false
       }
}

Параметры так же могут быть переданны в ссылке:

amqp[s]://[user:password@]hostname[:port][/vhost]

var conn =  amqp.createConnection({url: "amqp://guest:guest@localhost:5672"});

Вторым атрибутом передаём параметры AMQP. Значения по умолчанию:

{ defaultExchangeName: ''
, reconnect: true
, reconnectBackoffStrategy: 'linear'
, reconnectExponentialLimit: 120000
, reconnectBackoffTime: 1000
}

Если параметр reconnect включён, то драйвер попробует переподключиться.

connection.publish(queueName, body)

Отправляет сообщение в очередь с заданым именем, если по умолчанию DefaultExchange = ''.

connection.end()

Завершайте соеденение правильно!

Queue(очереди)

Очередь вызывает колбек который будет использован для работы с этой очередь.

var q = connection.queue('my-queue', function (queue) {
  console.log('Queue ' + queue.name + ' is open');
});

Если первый аргумент пуст, то сервер сам сгенерит рандомное имя.

connection.queue(name, options, openCallback)

Возвращает ссылку на очередь. Параметры:

passive: тип boolean, по умолчанию false. Если true, то сервер не будет создавать очереди. Служит чтобы проверить есть ли такая очередь
durable: тип boolean, по умолчанию false. Если true, то очереди остануться даже после перезагрузки сервера.
exclusive: тип boolean по умолчанию false. Эксклюзивные очереди могут использоваться только в текущем соеденении. Если true то подразумеваеться что включено авто удаление.
autoDelete: тип bollean по умолчанию true. Если установленно, очередь удалиться когда опустеет.
noDeclare: Тут не понял. boolean, default false. If set, the queue will not be declared, this will allow a queue to be deleted if you dont know its previous options.
arguments: Карта доп аргументов для передачи в очередь при её создании.
closeChannelOnUnsubscribe: Тип boolean, по умолчанию false. Закрыть канал после отписки.

queue.subscribe([options,] listener)

Тьфу ерунда - эта подписка. Вот как это работает:

q.subscribe(function (message, headers, deliveryInfo) {
  console.log('Got a message with routing key ' + deliveryInfo.routingKey);
});

Она автоматически будет подтверждать получение каждого сообщения.

Есть несколько вариантов получения сообщения. Первый: установите параметр {ack:true}(по умолчанию false) вы будите получать по одному сообщению за раз, если хотите получить следующие используйте q.shift(). Когда ack == false вы будете получать сообщения так быстро как сможете.

Вы так же можете использовать prefetchCount возможность увеличить окно получаемых сообщений, если оно равно 0 значит неограничено, по умолчанию равно еденице.