Implementing Publish-Subscribe Pattern

1. Creating Fanout Exchange

Property	Description	Default Value
name	Exchange name	""
type	Exchange type (fanout)	"fanout"
durable	Whether the exchange survives broker restarts	false
autoDelete	Whether the exchange is deleted when no longer in use	false

Example: Creating Fanout Exchange in RabbitMQ

const amqp = require('amqplib');

async function createFanoutExchange() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const exchangeName = 'logs';
  await channel.assertExchange(exchangeName, 'fanout', { durable: false });

  console.log(`Exchange ${exchangeName} created successfully`);
  await channel.close();
  await connection.close();
}

createFanoutExchange().catch(console.error);

Note: Fanout exchanges route messages to all queues bound to them, regardless of routing key.

2. Creating Temporary Queues

Property	Description	Default Value
queue	Name of the queue to declare	Empty string (auto-generated)
durable	Whether the queue survives broker restarts	`false`
exclusive	Whether the queue is deleted when the connection closes	`false`
autoDelete	Whether the queue is deleted when the last consumer unsubscribes	`false`

Example: Declaring a Temporary Queue in Java

import com.rabbitmq.client.*;

Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();

// Declare an exclusive, auto-delete queue
String queueName = channel.queueDeclare().getQueue();
System.out.println("Temporary queue created: " + queueName);

// Use the queue for publishing/consuming messages
channel.queueBind(queueName, "logs", "info");

// When done, the queue will be automatically deleted

Note: Temporary queues are useful for scenarios where you want to ensure that a queue is cleaned up after use, such as in testing or one-time message processing.

3. Binding Subscriber Queues

Binding Key	Queue Name	Exchange Type	Description
logs.#	`error-queue`	`topic`	Binds all logs with severity `error` or higher
logs.info	`info-queue`	`topic`	Binds only `info` level logs
logs.*	`all-logs`	`topic`	Binds all logs regardless of severity

Example: Binding Queues to Exchange

const amqp = require('amqplib');

async function bindQueues() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const exchangeName = 'logs';
  const queueName = 'error-queue';

  await channel.assertExchange(exchangeName, 'topic', { durable: true });
  await channel.assertQueue(queueName, { durable: true });

  // Bind queue to exchange with routing key
  await channel.bindQueue(queueName, exchangeName, 'logs.#');

  console.log('Queue bound to exchange');
  await channel.close();
  await connection.close();
}

bindQueues().catch(console.error);

Note: The binding key logs.# uses the # wildcard to match any number of words after logs..

4. Publishing to All Subscribers

Property	Description	Default Value	Notes
exchange.name	Name of the exchange to publish to	""	Must be unique within the broker
exchange.type	Type of exchange (fanout, direct, topic, headers)	fanout	fanout is used for publish-subscribe
routing.key	Used for routing messages in direct/topic exchanges	""	Ignored for fanout exchanges
message.body	Content of the message to publish	""	Can be any valid payload (text, JSON, binary)

Example: Publishing a Message to All Subscribers

const amqp = require('amqplib');

async function publishToAll() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  const exchangeName = 'notifications';
  const message = { text: 'Hello, subscribers!' };
  
  await channel.assertExchange(exchangeName, 'fanout', { durable: true });
  channel.publish(exchangeName, '', Buffer.from(JSON.stringify(message)));
  
  await channel.close();
  await connection.close();
}

publishToAll().catch(console.error);

Note: Fanout exchanges route messages to all bound queues, making them ideal for publish-subscribe patterns.

Key Points

Fanout exchanges are used for publish-subscribe messaging
All bound queues receive a copy of the message
No routing key is required for fanout exchanges
Messages are delivered reliably if queues are durable

5. Implementing Durable Subscriptions

Durable subscriptions in RabbitMQ ensure that messages are not lost even if the consumer or broker restarts. This is achieved through a combination of durable queues, persistent messages, and proper acknowledgment handling.

Component	Description	Configuration
durable queue	Queues that survive broker restarts	`true`
persistent message	Messages that survive broker restarts	`true`
manual acknowledgment	Consumer must explicitly acknowledge messages	`true`

Example: Setting up a durable subscription

const amqp = require('amqplib');

async function setupDurableSubscription() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  // Declare a durable queue
  await channel.assertQueue('my-durable-queue', { durable: true });

  // Set up consumer with manual acknowledgment
  await channel.consume('my-durable-queue', (msg) => {
    console.log("Received: ", msg.content.toString());
    channel.ack(msg); // Acknowledge the message
  }, { noAck: false });

  console.log("Durable subscription set up");
}

setupDurableSubscription().catch(console.error);

Note: Durable queues and persistent messages must be used together to ensure message durability across broker restarts.

Warning: If noAck: true is set, messages are automatically acknowledged and cannot be redelivered if the consumer fails.

6. Handling Subscriber Failures

Failure Type	Description	Handling Strategy	Recovery Mechanism
Message NACK	Subscriber explicitly rejects a message	Requeue or dead-letter queue	Message re-delivered to another subscriber or routed to DLX
Connection Failure	Subscriber drops connection during processing	Automatic reconnection or message redelivery	Broker retries delivery based on retry policy
Timeout	Subscriber does not acknowledge within timeout	Message requeue or DLX routing	Re-delivery after timeout or manual intervention

Example: Message Requeue on NACK

const amqp = require('amqplib');

async function handleMessages() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  channel.consume('myQueue', (msg) => {
    try {
      // Process message
      processMessage(msg.content);
      
      // Acknowledge successful processing
      channel.ack(msg);
    } catch (error) {
      console.error('Processing failed:', error);
      
      // Reject and requeue the message
      channel.nack(msg, false, true); // requeue = true
    }
  });
}

function processMessage(data) {
  // Simulate processing that might fail
  if (Math.random() < 0.2) throw new Error('Random failure');
  console.log('Processed:', data.toString());
}

Example: Dead Letter Exchange (DLX) Configuration

// Set up queue with DLX for failed messages
const queueOptions = {
  durable: true,
  arguments: {
    'x-dead-letter-exchange': 'dlx.exchange',
    'x-message-ttl': 30000 // 30 seconds TTL
  }
};

await channel.assertQueue('myQueue', queueOptions);
await channel.assertExchange('dlx.exchange', 'direct');
await channel.bindQueue('myQueue', 'dlx.exchange', 'failed-messages');

Note: When a message is rejected with requeue=true, it may be delivered to the same subscriber again. For better reliability, use a dead letter exchange (DLX) or implement circuit breaker logic to prevent infinite loops.

Key Recovery Strategies

Requeue on Nack: Return failed messages to queue for retry
DLX Routing: Route failed messages to error handling queue
Message TTL: Set timeouts to prevent indefinite retries
Connection Recovery: Use automatic reconnection for transient failures

7. Using Multiple Publishers

When implementing the publish-subscribe pattern with RabbitMQ, multiple publishers can send messages to the same exchange. Each publisher operates independently and does not need to know about other publishers or subscribers.

Property	Description	Example
Exchange Name	Name of the exchange to which messages are published	`logs`
Message Body	Content of the message being published	`"INFO: System started"`
Routing Key	Used to route messages to queues based on binding rules	`"info"`

Example: Publishing to Exchange

const amqp = require('amqplib');

async function publishMessage() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  const exchangeName = 'logs';
  const message = 'INFO: System started';
  const routingKey = 'info';

  await channel.assertExchange(exchangeName, 'topic', { durable: false });
  channel.publish(exchangeName, routingKey, Buffer.from(message));
  
  console.log(`[x] Sent '${message}'`);
  
  await channel.close();
  await connection.close();
}

publishMessage();

Note: Publishers do not need to know about the existence of queues or consumers. They only interact with exchanges.

8. Implementing Topic-Based Pub/Sub

Component	Description	Notes
Exchange Type	`topic`	Allows routing based on patterns with wildcards
Routing Key Pattern	`* (star)` and `# (hash)`	`*` matches one word, `#` matches zero or more words
Message Producer	Uses `basic_publish`	Must specify routing key for topic exchange
Consumer Binding	Uses `queue_bind`	Binding key can contain wildcards to match multiple topics

Example: Publishing a Message with Topic Routing

const amqp = require('amqplib');

async function publishMessage() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  const exchangeName = 'logs';
  const routingKey = 'error.system';
  
  await channel.assertExchange(exchangeName, 'topic', { durable: true });
  channel.publish(exchangeName, routingKey, Buffer.from('System error occurred'), {
    persistent: true
  });
  
  await channel.close();
  await connection.close();
}

publishMessage().catch(console.error);

Example: Consuming Messages with Topic Binding

const amqp = require('amqplib');

async function consumeMessages() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  const exchangeName = 'logs';
  const queueName = 'error_queue';
  
  await channel.assertExchange(exchangeName, 'topic', { durable: true });
  await channel.assertQueue(queueName, { durable: true });
  
  // Bind to all error messages
  await channel.bindQueue(queueName, exchangeName, 'error.#');
  
  channel.consume(queueName, (msg) => {
    console.log(`Received: ${msg.content.toString()}`);
    channel.ack(msg);
  }, { noAck: false });
}

consumeMessages().catch(console.error);

Topic Exchange Patterns

info.* matches any message with routing key starting with info.
#.error matches any message ending with .error
*.system.* matches any message with exactly one word between .system.

Note: Topic exchanges are more flexible than direct or fanout exchanges but can be less performant due to pattern matching.

9. Filtering Messages

Filter Type Description Usage Notes

Header Filters

Filter messages based on message headers

Filter Type	Description	Usage	Notes
Header Filters	Filter messages based on message headers	`const queue = await channel.assertQueue('', { exclusive: true }); await channel.bindQueue(queue.queue, 'exchange', '', { 'header1': 'value1' });`	Headers must match exactly
Routing Key Filters	Filter messages by routing key matching	`const queue = await channel.assertQueue('queue1'); await channel.bindQueue(queue.queue, 'direct_exchange', 'info');`	Used with direct exchanges
Topic Filters	Filter messages using wildcard patterns in routing keys	`const queue = await channel.assertQueue('queue1'); await channel.bindQueue(queue.queue, 'topic_exchange', '*.info');`	Supports `*` and `#` wildcards

const queue = await channel.assertQueue('', { exclusive: true });
await channel.bindQueue(queue.queue, 'exchange', '', {
  'header1': 'value1'
});

Headers must match exactly

Routing Key Filters

Filter messages by routing key matching

const queue = await channel.assertQueue('queue1');
await channel.bindQueue(queue.queue, 'direct_exchange', 'info');

Used with direct exchanges

Topic Filters

Filter messages using wildcard patterns in routing keys

const queue = await channel.assertQueue('queue1');
await channel.bindQueue(queue.queue, 'topic_exchange', '*.info');

Supports * and # wildcards

Example: Using Header Filters

const amqp = require('amqplib');
async function setupHeadersFilter() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();

  const exchangeName = 'headers_exchange';
  const queueName = 'filtered_queue';

  await channel.assertExchange(exchangeName, 'headers', { durable: false });
  const queue = await channel.assertQueue('', { exclusive: true });
  await channel.bindQueue(queue.queue, exchangeName, '', {
    'priority': 'high',
    'type': 'alert'
  });

  channel.publish(exchangeName, '', Buffer.from('High priority alert'), {
    headers: { priority: 'high', type: 'alert' }
  });

  await channel.close();
  await connection.close();
}

Note: Header filters require exact key-value matches. For more complex filtering logic, consider using a fanout exchange with message properties.

10. Monitoring Subscriber Count

Property	Description	Value
subscriber.count	Number of active subscribers to a topic/exchange	`0`
exchange.name	Name of the exchange being monitored	`my-topic-exchange`
queue.name	Name of the queue subscribed to the exchange	`subscriber-queue`

Example: Using RabbitMQ Management Plugin

curl -u guest:guest http://localhost:15672/api/exchanges/%2f/my-topic-exchange

Note: The subscriber.count is updated in real-time as subscribers connect or disconnect.

Warning: If monitoring via API, ensure that the management plugin is enabled and accessible.