我刚刚开始使用RabbitMQ和AMQP。
我有一个消息队列
我有多个消费者，我希望用相同的消息做不同的事情。
大多数RabbitMQ文档似乎都关注于循环，即单个消息由单个消费者消费，负载分布在每个消费者之间。这确实是我亲眼所见的行为。
举个例子:生产者只有一个队列，每2秒发送一次消息:
var amqp = require('amqp');
var connection = amqp.createConnection({ host: "localhost", port: 5672 });
var count = 1;
connection.on('ready', function () {
var sendMessage = function(connection, queue_name, payload) {
var encoded_payload = JSON.stringify(payload);
connection.publish(queue_name, encoded_payload);
}
setInterval( function() {
var test_message = 'TEST '+count
sendMessage(connection, "my_queue_name", test_message)
count += 1;
}, 2000)
})
这是一位消费者:
var amqp = require('amqp');
var connection = amqp.createConnection({ host: "localhost", port: 5672 });
connection.on('ready', function () {
connection.queue("my_queue_name", function(queue){
queue.bind('#');
queue.subscribe(function (message) {
var encoded_payload = unescape(message.data)
var payload = JSON.parse(encoded_payload)
console.log('Recieved a message:')
console.log(payload)
})
})
})
如果我启动消费者两次，我可以看到每个消费者都在循环行为中使用备用消息。例如，我将在一个终端上看到消息1,3,5，在另一个终端上看到消息2,4,6。
我的问题是:
我能让每个消费者收到相同的消息吗?也就是说，两个消费者都得到消息1,2,3,4,5,6 ?这在AMQP/RabbitMQ中被称为什么?它通常是如何配置的?
这种做法普遍吗?我是否应该让交换器将消息路由到两个单独的队列中，只有一个消费者?
我能让每个消费者收到相同的消息吗?也就是说，两个消费者都得到消息1,2,3,4,5,6 ?这在AMQP/RabbitMQ中被称为什么?它通常是如何配置的?
不，如果消费者在同一个队列上，就不会。摘自RabbitMQ的AMQP概念指南:
重要的是要理解，在AMQP 0-9-1中，消息在消费者之间是负载均衡的。
这似乎意味着队列中的循环行为是给定的，不可配置。也就是说，为了让多个使用者处理相同的消息ID，需要单独的队列。
这种做法普遍吗?我是否应该让交换器将消息路由到两个单独的队列中，只有一个消费者?
不，它不是，单个队列/多个消费者，每个消费者处理相同的消息ID是不可能的。让交换器将消息路由到两个单独的队列确实更好。
由于我不需要太复杂的路由，扇出交换机将很好地处理这个问题。我之前没有过多关注exchange，因为node-amqp具有“默认交换”的概念，允许您直接将消息发布到连接，但是大多数AMQP消息都发布到特定的交换。
以下是我的扇出交换，包括发送和接收:
var amqp = require('amqp');
var connection = amqp.createConnection({ host: "localhost", port: 5672 });
var count = 1;
connection.on('ready', function () {
connection.exchange("my_exchange", options={type:'fanout'}, function(exchange) {
var sendMessage = function(exchange, payload) {
console.log('about to publish')
var encoded_payload = JSON.stringify(payload);
exchange.publish('', encoded_payload, {})
}
// Recieve messages
connection.queue("my_queue_name", function(queue){
console.log('Created queue')
queue.bind(exchange, '');
queue.subscribe(function (message) {
console.log('subscribed to queue')
var encoded_payload = unescape(message.data)
var payload = JSON.parse(encoded_payload)
console.log('Recieved a message:')
console.log(payload)
})
})
setInterval( function() {
var test_message = 'TEST '+count
sendMessage(exchange, test_message)
count += 1;
}, 2000)
})
})
RabbitMQ / AMQP:单队列，同一消息和页面刷新的多个消费者。
rabbit.on('ready', function () { });
sockjs_chat.on('connection', function (conn) {
conn.on('data', function (message) {
try {
var obj = JSON.parse(message.replace(/\r/g, '').replace(/\n/g, ''));
if (obj.header == "register") {
// Connect to RabbitMQ
try {
conn.exchange = rabbit.exchange(exchange, { type: 'topic',
autoDelete: false,
durable: false,
exclusive: false,
confirm: true
});
conn.q = rabbit.queue('my-queue-'+obj.agentID, {
durable: false,
autoDelete: false,
exclusive: false
}, function () {
conn.channel = 'my-queue-'+obj.agentID;
conn.q.bind(conn.exchange, conn.channel);
conn.q.subscribe(function (message) {
console.log("[MSG] ---> " + JSON.stringify(message));
conn.write(JSON.stringify(message) + "\n");
}).addCallback(function(ok) {
ctag[conn.channel] = ok.consumerTag; });
});
} catch (err) {
console.log("Could not create connection to RabbitMQ. \nStack trace -->" + err.stack);
}
} else if (obj.header == "typing") {
var reply = {
type: 'chatMsg',
msg: utils.escp(obj.msga),
visitorNick: obj.channel,
customField1: '',
time: utils.getDateTime(),
channel: obj.channel
};
conn.exchange.publish('my-queue-'+obj.agentID, reply);
}
} catch (err) {
console.log("ERROR ----> " + err.stack);
}
});
// When the visitor closes or reloads a page we need to unbind from RabbitMQ?
conn.on('close', function () {
try {
// Close the socket
conn.close();
// Close RabbitMQ
conn.q.unsubscribe(ctag[conn.channel]);
} catch (er) {
console.log(":::::::: EXCEPTION SOCKJS (ON-CLOSE) ::::::::>>>>>>> " + er.stack);
}
});
});
