Implementing Single Active Consumer

1. Enabling Single Active Consumer

The Single Active Consumer (SAC) pattern in RabbitMQ ensures that only one consumer processes messages from a queue at any given time, even if multiple consumers are connected to the same queue. This is particularly useful for scenarios where message ordering or exclusive processing is required.

Property	Value	Description
single_active_consumer	`true`	Enables the Single Active Consumer feature on a queue.
queue.name	`my-queue`	Name of the queue where SAC is enabled.
consumer.tag	`sac-consumer-1`	Unique identifier for the consumer.

Example: Enabling SAC in a Queue

const amqp = require('amqplib');

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

  const queueName = 'my-queue';
  const options = {
    durable: true,
    arguments: {
      'x-single-active-consumer': true
    }
  };

  await channel.assertQueue(queueName, options);
  console.log('Queue with SAC enabled created.');
}

setupQueueWithSAC().catch(console.error);

Note: The x-single-active-consumer argument must be set when declaring the queue. It is not possible to enable SAC on an existing queue without redeclaring it.

Warning: If multiple consumers are connected to a queue with SAC enabled, only one will receive messages at any time. The active consumer is determined by the broker and can change if a consumer disconnects or becomes unresponsive.

2. Understanding Consumer Failover

Consumer failover in RabbitMQ ensures that if one consumer in a group becomes unavailable, another consumer takes over its responsibilities. This is particularly useful for ensuring high availability and preventing message loss in distributed systems.

Feature	Description	Notes
single active consumer	Ensures only one consumer processes messages at a time from a queue.	Used in scenarios where message ordering or state consistency is required.
failover mechanism	Automatically switches to a backup consumer when the active one fails.	Requires proper queue and consumer configuration to enable.
consumer state tracking	RabbitMQ tracks consumer states to determine active vs. inactive consumers.	Uses internal state management and heartbeats for monitoring.

Example: Consumer Failover Configuration

const amqp = require('amqplib');

async function setupConsumer() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  const queueName = 'my-queue';
  const consumerTag = 'consumer-1';

  // Declare queue with single active consumer
  await channel.assertQueue(queueName, { durable: true });
  
  // Start consuming with single active consumer
  await channel.consume(queueName, (msg) => {
    console.log('Received:', msg.content.toString());
    channel.ack(msg);
  }, { consumerTag, noAck: false });
  
  console.log('Consumer started with failover support');
}

setupConsumer().catch(console.error);

Note: The consumerTag must be unique per consumer to enable failover. If multiple consumers share the same tag, they may interfere with each other's state tracking.

Key Points for Failover

Single active consumer ensures only one consumer processes messages at a time.
Failover occurs when the active consumer disconnects or becomes unresponsive.
Consumers must be configured with unique consumerTag to support failover.
Message ordering is preserved during failover transitions.

3. Using with Ordered Processing

Property	Description	Values
single_active_consumer	Enables single active consumer mode for a queue	`true`

Example: Declaring a Queue with Single Active Consumer

const queueOptions = {
  durable: true,
  arguments: {
    'x-single-active-consumer': true
  }
};

channel.assertQueue('ordered-queue', queueOptions);

Note: When x-single-active-consumer is enabled, only one consumer can be active at a time. If a new consumer connects while another is active, it will be queued until the current consumer disconnects.

Warning: Enabling x-single-active-consumer can introduce bottlenecks if the consumer is slow or experiences delays.

Key Points

Ensures only one consumer processes messages from a queue at a time.
Useful for ordered processing where message sequence matters.
Consumer must explicitly handle reconnection logic to resume processing.

4. Combining with Prefetch

Property	Description	Default Value	Notes
prefetchCount	Number of messages to prefetch per consumer	`1`	Controls how many messages are delivered to a consumer before acknowledging one
prefetchSize	Maximum size of prefetch in bytes	`0`	Zero means no limit; can be used to control memory usage
global	Whether prefetch is applied globally or per-consumer	`false`	When `true`, applies to all consumers on the channel

Example: Setting Prefetch Count

const amqp = require('amqplib');

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

  // Set prefetch count to 1 for single active consumer behavior
  channel.prefetch(1);

  await channel.assertQueue('task_queue', { durable: true });
  await channel.consume('task_queue', (msg) => {
    console.log("Processing message", msg.content.toString());
    channel.ack(msg);
  }, { noAck: false });
}

Note: Setting prefetchCount to 1 ensures that only one message is delivered at a time to a consumer, which supports the single active consumer pattern.

Warning: If prefetchCount is set higher than 1, multiple consumers may receive messages simultaneously, which can break the single active consumer guarantee.

5. Monitoring Active Consumer

Example: Consumer Metrics

// Example of monitoring consumer metrics in RabbitMQ
const amqp = require('amqplib');

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

  const queueName = 'my-queue';
  await channel.assertQueue(queueName, { durable: true });

  // Consumer metrics can be accessed via management plugin
  const consumerTag = await channel.consume(queueName, (msg) => {
    console.log('Received message:', msg.content.toString());
    channel.ack(msg);
  });

  // Log consumer tag for monitoring
  console.log('Consumer Tag:', consumerTag.consumerTag);
}

monitorConsumer().catch(console.error);

Property	Description	Values
consumerTag	Unique identifier for the consumer	`amq.ctag-...`
messageCount	Number of messages in the queue	`0` to `∞`
consumerCount	Number of active consumers on the queue	`0` to `∞`

Note: The consumerTag is essential for identifying which consumer is active in a Single Active Consumer setup.

Warning: If multiple consumers are active, only one will process messages. Ensure monitoring tools track consumerTag to detect active consumers.

Key Monitoring Points

Track consumerTag to identify the active consumer.
Monitor messageCount to understand queue load.
Check consumerCount to ensure only one consumer is active.

6. Handling Consumer Failures

Example: Consumer Failure with Manual Acknowledgement

const amqp = require('amqplib');

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

  const queue = 'task_queue';
  await channel.assertQueue(queue, { durable: true });

  // Set up consumer with manual acknowledgment
  channel.consume(queue, async (msg) => {
    try {
      // Simulate processing
      await processTask(msg.content.toString());
      
      // Acknowledge message if successful
      channel.ack(msg);
    } catch (error) {
      console.error('Consumer failed:', error);
      
      // Reject message and requeue for retry
      channel.nack(msg, false, true);
    }
  }, { noAck: false });
}

async function processTask(task) {
  // Simulate task processing that might fail
  if (Math.random() < 0.2) throw new Error('Random failure');
  console.log('Processing task:', task);
}

Property	Description	Default
channel.nack(msg, requeue, allUpTo)	Rejects a message and optionally requeues it	false, false
channel.ack(msg)	Acknowledges a message as successfully processed	—
channel.reject(msg, requeue)	Rejects a message and optionally requeues it	false

Note: When using channel.nack(msg, false, true), the allUpTo parameter can be used to reject all messages up to and including the current one.

Consumer Failure Handling Strategy

Manual Acknowledgement: Consumers must explicitly acknowledge messages to prevent re-delivery.
Error Recovery: If a consumer fails, it should reject the message with nack or reject to allow reprocessing.
Requeue Option: Setting requeue to true allows the message to be requeued and picked up by another consumer.

7. Understanding Limitations

Key Limitations of Single Active Consumer Pattern

Consumer failover is not automatic; requires manual intervention or external monitoring
Does not support dynamic scaling of consumers within a queue
May cause performance bottlenecks if the single active consumer becomes overwhelmed
Does not provide true load balancing across multiple consumers
Requires careful handling of consumer lifecycle to avoid message loss or duplication

Limitation	Description	Impact
Failover Complexity	When the active consumer crashes or disconnects, another consumer must take over. This process is not automatic and requires external monitoring or health checks to detect failure.	Increased risk of message loss or processing delays
Scalability Constraints	The pattern does not support scaling consumers dynamically. If the workload increases, additional consumers cannot be added without modifying queue configuration or consumer logic.	Potential performance degradation under load
Resource Contention	Only one consumer processes messages at a time, which can lead to resource contention and reduced throughput if the consumer is CPU or I/O bound.	Reduced parallelism and efficiency
Message Ordering	While it ensures sequential processing, it does not guarantee strict ordering across multiple consumers or queues.	Potential for out-of-order message handling in distributed systems

Note: The Single Active Consumer pattern is best suited for scenarios where strict message ordering and exclusive processing are more important than horizontal scaling or high throughput.

8. Using with Quorum Queues

Quorum queues in RabbitMQ provide high availability and durability by replicating messages across multiple nodes. When using Single Active Consumer (SAC) with quorum queues, the consumer is activated only on one node at a time, ensuring that only one consumer processes messages from the queue.

Configuration	Description	Notes
x-single-active-consumer	Enables Single Active Consumer for a queue	Set to `true` to activate SAC
x-queue-type	Queue type	Must be set to `quorum` for quorum queues

Example: Declaring a Quorum Queue with SAC

const queueOptions = {
  durable: true,
  autoDelete: false,
  arguments: {
    'x-queue-type': 'quorum',
    'x-single-active-consumer': true
  }
};

await channel.assertQueue('my-quorum-queue', queueOptions);

Note: When using SAC with quorum queues, the consumer is automatically activated on one node and deactivated on others. If the active consumer fails, another node will take over.

Warning: SAC does not guarantee that all messages are processed in order when using quorum queues. The order is maintained within the consumer node, but messages may arrive in different orders across nodes.