PROSAGA码农传奇-rabbitmq-RabbitMQ / AMQP：单个队列，同一条消息的多个使用者？

RabbitMQ / AMQP：单个队列，同一条消息的多个使用者？

作者: 狗头军师
发布时间: 2024-12-19 04:51:37 (2月前)

我一般只是开始使用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) }) ``` 这是一个消费者：
<pre><code>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)
 })
 })
})
</code></pre>如果我启动使用者两次，则可以看到每个使用者都以循环方式使用替代消息。例如，我将在一个终端中看到消息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是不可能的。让交换将消息路由到两个单独的队列中确实更好。

由于我不需要太复杂的路由，因此扇出交换机可以很好地解决这一问题。由于node-amqp具有“默认交换”的概念，允许您直接将消息发布到连接中，因此我没有过多关注Exchange，但是大多数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) 
 })
})
```