我想使用来自 rabbitmq 服务的消息,并且对于我收到的每条消息,我都想做一些事情(例如:将该消息放入数据库,处理消息并通过 RabbitMq 通过另一个队列发送回复)每条消息。
目前我的 RabbitMq 消费者代码如下:
const all = require('bluebird').all;
const basename = require('path').basename;
function receive() {
const severities = process.argv.slice(2);
if (severities.length < 1) {
console.warn('Usage: %s [info] [warning] [error]',
basename(process.argv[1]));
process.exit(1);
}
let config = {
protocol: 'amqp',
hostname: 'localhost',
port: 5672,
username: 'rumesh',
password: 'password',
locale: 'en_US',
frameMax: 0,
heartbeat: 0,
vhost: '/',
};
amqp.connect(config).then(function (conn) {
process.once('SIGINT', function () {
conn.close();
});
return conn.createChannel().then(function (ch) {
let queue = 'test';
let exchange = 'test-exchange';
let key = 'python-key';
let exchange_type = 'direct';
let ok = ch.assertExchange(exchange, exchange_type, {durable: true});
ok = ok.then(function () {
return ch.assertQueue(queue, { durable: true});
});
ok = ok.then(function (qok) {
const queue = qok.queue;
return all(severities.map(function (sev) {
ch.bindQueue(queue, exchange, sev,{durable: true});
})).then(function () {
return queue;
});
});
ok = ok.then(function (queue) {
return ch.consume(queue, logMessage, {noAck: true});
});
return ok.then(function () {
console.log(' [*] Waiting for logs. To exit press CTRL+C.');
});
function logMessage(msg) {
console.log(" [x] %s:'%s'",
msg.fields.routingKey,
msg.content.toString());
}
});
}).catch(console.warn);
}
module.exports = receive;```