Implementing Multiple Consumers Receiving the Same Message in RabbitMQ

Default Message Distribution in RabbitMQ

In the AMQP protocol and RabbitMQ implementation, when multiple consumers connect to the same queue, the system defaults to round-robin message distribution. This means each message is received and processed by only one consumer, with the system balancing message load across all active consumers.

The following code demonstrates typical round-robin behavior:

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('Received a message:')
      console.log(payload)
    })
  })
})

When starting two instances of such consumers, messages 1, 3, 5 are received by the first consumer while messages 2, 4, 6 go to the second consumer. This mechanism ensures load balancing but prevents multiple consumers from processing the same message simultaneously.

Broadcast Mechanism with Fanout Exchange

To enable multiple consumers to receive identical messages, a different architectural approach is required. The fanout exchange provides an ideal solution by broadcasting every received message to all bound queues.

Here's the complete implementation using a fanout exchange:

var amqp = require('amqp');
var connection = amqp.createConnection({ host: "localhost", port: 5672 });
var count = 1;

connection.on('ready', function () {
  connection.exchange("my_exchange", {type: 'fanout'}, function(exchange) {   
    var sendMessage = function(exchange, payload) {
      console.log('about to publish')
      var encoded_payload = JSON.stringify(payload);
      exchange.publish('', encoded_payload, {})
    }

    // Queue configuration for message reception
    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('Received a message:')
        console.log(payload)
      })
    })
  
    setInterval(function() {    
      var test_message = 'TEST ' + count
      sendMessage(exchange, test_message)  
      count += 1;
    }, 2000) 
  })
})

In this implementation, messages published to the fanout exchange are automatically replicated to all bound queues. Each queue can have its own consumer, enabling multiple consumers to independently process identical messages.

Architectural Design and Best Practices

The fanout exchange architecture offers several advantages:

• Producer-Consumer Decoupling: Producers don't need to know about the number or types of consumers
• Flexible Scaling: New consumer queues can be added without impacting existing systems
• Independent Processing: Each consumer can process messages at its own pace without interference

Alternative approaches like using amq.direct exchange with identical routing keys can achieve similar results, but fanout exchanges provide a more concise and efficient solution for broadcast scenarios.

In practical applications, the choice of pattern should align with specific business requirements. For scenarios requiring each message to be processed by multiple independent systems, fanout exchanges with multiple queue bindings represent the best practice approach.