我有以下出版商和订阅者:
// publisher.js
var amqp = require('amqplib/callback_api');
amqp.connect('amqp://localhost', function(err, conn) {
conn.createChannel(function(err, ch) {
var ex = 'logs';
var msg = process.argv.slice(2).join(' ') || 'Hello World!';
ch.assertExchange(ex, 'fanout', {durable: true});
ch.publish(ex, '', new Buffer(msg));
console.log(" [x] Sent %s", msg);
});
setTimeout(function() { conn.close(); process.exit(0) }, 500);
});
//subscriber.js
var amqp = require('amqplib/callback_api');
amqp.connect('amqp://localhost', function(err, conn) {
conn.createChannel(function(err, ch) {
var ex = 'logs';
ch.assertExchange(ex, 'fanout', {durable: true});
ch.assertQueue('', {exclusive: true}, function(err, q) {
console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", q.queue);
ch.bindQueue(q.queue, ex, '');
ch.consume(q.queue, function(msg) {
if(msg.content) {
console.log(" [x] %s", msg.content.toString());
}
}, {noAck: true});
});
});
});
如果我运行Advertiser.js发布两条消息,然后运行subscriber.js,我希望订阅者打印我过去发布的两条消息,但实际上订阅者没有收到任何消息。
如何让RabbitMQ持久化已发布的消息,允许任何订阅者从一开始或在他们上次消费的任何地方消费?
1条答案
按热度按时间omtl5h9j1#
如果RabbitMQ无法将消息路由到队列,则该消息将被丢弃。
由于您在订阅者代码之前运行发布者代码,因此没有队列绑定到您的交换机,因此没有地方可以路由消息。
您需要在发布者之前运行订阅者代码,或者必须让发布者或其他进程创建命名队列并将其绑定到交换。然后,您的订阅者将使用该队列。