RabbitMQ 消费者应该通过路由键而不是 QueueName 来监听

Question

我使用以下代码示例创建了一个发布者。

var amqp = require('amqplib/callback_api');

amqp.connect('amqp://localhost', function(err, conn) {
    conn.createChannel(function(err, ch) {
        var args = process.argv.slice(2);
        var routingkey = (args.length > 0) ? args[0] : 'anonymous.info';
        var keys = routingkey.split(".")

        var exchange = keys[0];
        queueName = keys[1];
        var msg = args.slice(1).join(' ') || 'Hello World!';

        console.log("Exchange---------------------" , exchange);
        console.log("routingkey-------------------", routingkey);
        console.log("queueName-------------------" , queueName);

        queueName = exchange;

        ch.assertExchange(exchange, 'topic', {durable: false});
        ch.assertQueue(queueName, {exclusive: false, durable: false}, false);

        ch.bindQueue(queueName, exchange, routingkey);

        ch.publish(exchange, routingkey, new Buffer(msg));
        console.log(" [x] Sent %s: '%s'", routingkey, msg);
    });

    setTimeout(function() { conn.close(); process.exit(0) }, 500);
});

基于上述发布者：

exchange : notification ::: 
 queueName : notification ::: 
 routingkey : ['notification.addworker' , 'notification.getworker',.....]

因此，我能够使用与多个路由键的绑定将消息发布到唯一的队列名称。

这是我的问题，我无法根据路由键使用消息，因为消息在发布时绑定到队列。

建议我上面的代码是否做错了。

消费者代码：

 var args = process.argv.slice(2);
 amqp.connect('amqp://localhost', function(err, conn) {
     conn.createChannel(function(err, ch) {
         var ex = 'notification';
         var args = process.argv.slice(2);
         var routingkey = (args.length > 0) ? args[0] : 'anonymous.info';
         var keys = routingkey.split(".")
         var exchange = keys[0];
         var queueName = exchange;
         ch.assertExchange(exchange, 'direct', {durable: false});

         ch.assertQueue(exchange, {exclusive: false, durable:false}, function(err, q) {
             console.log(' [*] Waiting for logs. To exit press CTRL+C');
             ch.bindQueue(queueName, ex, routingkey);
             console.log("Exchange---------------------" , exchange);
             console.log("routingkey-------------------", routingkey);
             console.log("queueName-------------------" , queueName);
             ch.consume(queueName, function(msg) {
                 console.log(" [x] %s:'%s'", msg.fields.routingKey, msg.content.toString());
             }, {noAck: true});
         });
     });
 });

消费者无法根据路由键过滤消息。

寻找我做错的地方的建议。我坚持使用主题。

感谢您的耐心等待。

Answer 1

我知道这是一篇旧帖子，但如果你对我的 2 美分感兴趣......我所做的是在发布命令中添加一些回复信息：

ch.publish(exchange, send_key, new Buffer(message), { correlationId: corrId, replyTo: q.queue });

其中 q.queue 来自 assetQueue 回调，可以指定回复队列。

然后就可以食用了...

ch.consume(q.queue, function(msg) {...}

不确定这是否准确回答了你的问题，我还没有设法消除队列名称，我自己还是一个初学者。