Consuming Messages

1. Creating Consumer

Property	Description	Default Value
queue	Name of the queue to consume messages from	None
noAck	If true, messages are automatically acknowledged after receiving	false
exclusive	If true, the consumer is exclusive to the queue	false
priority	Priority of the consumer (higher numbers are processed first)	None

Example: Creating a Basic Consumer

const amqp = require('amqplib');

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

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

createConsumer().catch(console.error);

2. Setting Consumer Tag

Property	Value	Description
consumerTag	`string`	Unique identifier for the consumer. If not provided, RabbitMQ generates one.
noLocal	`boolean`	If true, the server will not deliver messages published by this consumer to itself.
exclusive	`boolean`	If true, the queue will be exclusive to this consumer and will be deleted when the connection closes.
arguments	`object`	Additional arguments for the consumer (e.g., `{ "x-priority": 10 }`).

Example: Setting Consumer Tag

const amqp = require('amqplib');

async function consumeMessages() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  const queue = 'my-queue';

  await channel.assertQueue(queue, { durable: true });

  // Setting a custom consumer tag
  const consumerTag = 'my-consumer-123';
  await channel.consume(queue, (msg) => {
    console.log('Received:', msg.content.toString());
    channel.ack(msg);
  }, { consumerTag });

  console.log('Consumer started with tag:', consumerTag);
}

consumeMessages().catch(console.error);

Note: Consumer tags are useful for identifying consumers in logs and for managing consumer lifecycle.

3. Using Auto-Ack Mode

In auto-ack mode, RabbitMQ automatically acknowledges messages as soon as they are delivered to the consumer. This simplifies message processing but can lead to message loss if the consumer fails before processing.

Mode	Description	Use Case
autoAck = true	Messages are acknowledged immediately upon delivery.	When message loss is acceptable, and simplicity is preferred.
autoAck = false	Manual acknowledgment required before message is removed from queue.	When reliability is critical and messages must not be lost.

Example: Consuming with Auto-Ack

const amqp = require('amqplib');

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

  await channel.assertQueue(queue, { durable: true });
  
  // Consume with autoAck = true
  channel.consume(queue, (msg) => {
    console.log("Received: ", msg.content.toString());
    // Message is automatically acknowledged here
  }, { noAck: true });

  console.log("Waiting for messages...");
}

consumeWithAutoAck().catch(console.error);

Note: When noAck: true, the consumer does not need to explicitly acknowledge messages. This is equivalent to setting autoAck = true.

Warning: If a consumer crashes or disconnects while processing a message in auto-ack mode, that message is lost because it was already acknowledged.

4. Using Manual Ack Mode

Manual acknowledgment mode gives you fine-grained control over message processing by requiring explicit acknowledgment of each message.

Property	Description	Default Value
manualAck	Enables or disables manual acknowledgment mode for a consumer.	`false`

Example: Setting Up Manual Ack Mode

const amqp = require('amqplib');

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

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

  // Consume with manual acknowledgment
  channel.consume(queue, (msg) => {
    console.log('Received message:', msg.content.toString());
    
    // Process the message here
    // ...

    // Manually acknowledge the message
    channel.ack(msg);
  }, { noAck: false }); // Important: noAck must be false for manual ack
}

setupConsumer().catch(console.error);

Note: When manualAck is enabled, you must explicitly call channel.ack(msg) or channel.nack(msg) to acknowledge or reject the message. If not acknowledged, the message will be re-queued and potentially redelivered.

Warning: Failing to acknowledge messages in manual mode can lead to message loss or indefinite redelivery, depending on your broker configuration.

5. Getting Single Message

Property	Description	Default Value	Notes
messageId	Unique identifier for the message	Generated automatically	Used for deduplication and tracking
body	Content of the message	Empty string	Can be any valid payload (text, JSON, binary)
deliveryMode	Defines if message is persistent or transient	2 (persistent)	1 = transient, 2 = persistent
contentType	MIME type of the message body	text/plain	Used for content negotiation
timestamp	Timestamp when message was published	Current timestamp	Helps in tracking message age

Example: Fetching a Single Message

const amqp = require('amqplib');

async function getSingleMessage() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  const queue = 'my-queue';
  const msg = await channel.get(queue, { noAck: true });
  
  if (msg) {
    console.log('Message body:', msg.content.toString());
    console.log('Message ID:', msg.properties.messageId);
  } else {
    console.log('No messages in queue');
  }
  
  await channel.close();
  await connection.close();
}

getSingleMessage().catch(console.error);

Note: The noAck option determines whether the message is acknowledged after retrieval. Setting it to true removes the message from the queue immediately.

Warning: Using channel.get() is not suitable for high-throughput scenarios as it blocks the channel until a message is available.

6. Setting No-Local Flag

The no-local flag is used in RabbitMQ's basic.consume method to indicate that the broker should not deliver messages to the same connection that published them.

Property	Description	Values
no-local	When set to `true`, the broker will not deliver messages that were published by the same connection.	`true` or `false`

Example: Using no-local Flag

const amqp = require('amqplib');

async function consumeWithNoLocal() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  const queue = 'my-queue';

  await channel.assertQueue(queue, { durable: true });

  // Consume with no-local flag set to true
  channel.consume(queue, (msg) => {
    console.log('Received message:', msg.content.toString());
    channel.ack(msg);
  }, { noLocal: true });
}

consumeWithNoLocal().catch(console.error);

Note: The no-local flag is particularly useful in scenarios where a consumer is both publishing and consuming from the same queue, to avoid processing its own messages.

7. Setting Exclusive Consumer

Property	Value	Description
exclusive	`true` or `false`	When set to `true`, the consumer is exclusive to a queue and no other consumers can be created on that queue.

Example: Setting an Exclusive Consumer

const amqp = require('amqplib');

async function setExclusiveConsumer() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  const queue = 'my-queue';

  await channel.assertQueue(queue, { durable: true });
  
  // Setting an exclusive consumer
  const consumerTag = await channel.consume(queue, (msg) => {
    console.log('Received:', msg.content.toString());
    channel.ack(msg);
  }, { exclusive: true });

  console.log('Exclusive consumer tag:', consumerTag);
}

setExclusiveConsumer().catch(console.error);

8. Using Consumer Priorities

Property	Description	Default Value	Notes
x-priority	Defines the priority of the consumer	`0`	Higher values indicate higher priority

Example: Setting Consumer Priority

const amqp = require('amqplib');

async function setupConsumerWithPriority() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  const queueName = 'priority-queue';
  const priority = 5;

  await channel.assertQueue(queueName, {
    durable: true,
    arguments: {
      'x-max-priority': 10
    }
  });

  await channel.consume(queueName, (msg) => {
    console.log(`Received message with priority: ${msg.properties.priority}`);
    channel.ack(msg);
  }, {
    priority: priority
  });

  console.log(`Consumer with priority ${priority} registered`);
}

setupConsumerWithPriority().catch(console.error);

Note: Consumer priorities are only effective when the queue is configured with x-max-priority argument.

Warning: Consumer priorities do not guarantee message order, only priority-based processing.

9. Implementing Consumer Callbacks

Callback	Description	Parameters	Notes
onMessage	Called when a message is received from the queue	`message`	Process the message content and acknowledge it
onError	Called when an error occurs during message consumption	`error`	Handle connection or processing errors gracefully
onCancel	Triggered when the consumer is cancelled by the broker	`reason`	Reconnect or redeclare the consumer if needed

Example: Basic Message Processing

const amqp = require('amqplib');

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

  await channel.assertQueue(queue, { durable: true });

  channel.consume(queue, (msg) => {
    console.log("Received: ", msg.content.toString());
    channel.ack(msg);
  }, { noAck: false });
}

consumeMessages().catch(console.error);

Example: Error Handling in Callbacks

channel.consume(queue, (msg) => {
  try {
    const data = JSON.parse(msg.content.toString());
    processMessage(data);
    channel.ack(msg);
  } catch (err) {
    console.error("Error processing message: ", err);
    channel.nack(msg, false, false); // Reject and don't requeue
  }
}, { noAck: false });

10. Cancelling Consumer

Property	Description	Values
consumerTag	Identifier for the consumer	`string`
nowait	True if no reply is expected	`boolean`

Example: Cancel Consumer

const cancelConsumer = async () => {
  const consumerTag = 'consumer-123';
  try {
    await channel.cancel(consumerTag, false);
    console.log('Consumer cancelled successfully');
  } catch (error) {
    console.error('Failed to cancel consumer:', error);
  }
};

Note: The cancel method removes the consumer from the broker and stops message delivery.

11. Handling Consumer Cancellation

Property	Description	Default Value
consumer.cancel	Indicates if the consumer has been cancelled by the broker	false
consumer.tag	Unique identifier for the consumer	Generated by broker

Example: Consumer Cancellation Handling

const amqp = require('amqplib');

async function handleCancellation() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  const queue = 'my-queue';
  const consumerTag = await channel.consume(queue, (msg) => {
    if (msg && msg.fields.consumerTag) {
      console.log('Consumer tag:', msg.fields.consumerTag);
    }
    
    if (msg && msg.fields.redelivered) {
      console.log('Message redelivered');
    }
    
    channel.ack(msg);
  }, { noAck: false });
  
  // Cancel consumer if needed
  channel.cancel(consumerTag);
}

handleCancellation().catch(console.error);

Note: When a consumer is cancelled, the channel.cancel() method should be called to clean up resources.

Warning: If a consumer is cancelled, any unacknowledged messages will be requeued by the broker.

12. Monitoring Active Consumers

Property	Description	Values
consumer_tag	Identifier for the consumer	`amq.ctag-...`
queue	Name of the queue being consumed from	`my-queue`
prefetch_count	Number of messages to prefetch	`1`
active	Indicates if the consumer is active	`true`

Example: Get Active Consumers via Management Plugin

curl -u guest:guest http://localhost:15672/api/consumers

Example: Consumer Details in Management UI

{
  "consumer_tag": "amq.ctag-1234567890",
  "queue": {
    "name": "my-queue",
    "vhost": "/"
  },
  "prefetch_count": 1,
  "active": true
}