Skip to main content
Mapagam

Understanding Message Acknowledgments

1. Sending Basic Ack

Property Description Default Value
basic.ack Used to acknowledge a message received from a queue false
delivery_tag Identifier for the message being acknowledged None
multiple If true, acknowledges all messages up to and including the specified delivery tag false

Example: Sending Basic Ack

const amqp = require('amqplib');

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

  const queue = 'my-queue';
  const msg = { id: 1, content: 'Hello World' };

  await channel.assertQueue(queue, { durable: true });
  await channel.sendToQueue(queue, Buffer.from(JSON.stringify(msg)), { persistent: true });

  // Simulate receiving a message
  const consumerTag = await channel.consume(queue, (msg) => {
    console.log('Received:', msg.content.toString());
    
    // Acknowledge the message
    channel.ack(msg);
  });

  console.log('Message acknowledged');
}

sendBasicAck().catch(console.error);

2. Sending Basic Nack

The basic.nack method in RabbitMQ is used to negatively acknowledge a message, indicating that the consumer was unable to process the message successfully. This is particularly useful in scenarios where a message needs to be re-queued or sent to a dead-letter exchange.

Field Description Values
delivery_tag Identifier for the message being negatively acknowledged uint64
requeue Indicates whether the message should be re-queued or not true or false

Example: Using basic.nack to requeue a message

const amqp = require('amqplib');

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

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

  channel.consume(queue, (msg) => {
    if (msg.content.toString() === 'error') {
      // Re-queue the message
      channel.nack(msg, false, true);
    } else {
      // Accept the message
      channel.ack(msg);
    }
  });
}

nackExample().catch(console.error);
Note: The requeue parameter in basic.nack controls whether the message should be re-queued or sent to a dead-letter exchange. If set to false, the message is not re-queued and can be routed to a dead-letter exchange if configured.

3. Sending Basic Reject

When a consumer receives a message from RabbitMQ, it can either ack (accept) or nack (reject) the message. The basic.nack method allows a consumer to reject a message and optionally requeue it.

Field Description Values
delivery_tag Unique identifier for the message being rejected uint64
requeue Whether to requeue the message for redelivery true or false

Example: Rejecting a Message

const amqp = require('amqplib');

async function rejectMessage() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  const queue = 'my-queue';
  await channel.assertQueue(queue, { durable: true });
  
  channel.consume(queue, (msg) => {
    if (msg.content.toString() === 'reject') {
      // Reject and requeue the message
      channel.nack(msg, true);
    } else {
      // Accept the message
      channel.ack(msg);
    }
  });
}

rejectMessage().catch(console.error);
Note: If requeue is set to false, the message will be discarded or sent to a dead-letter exchange if configured.

4. Using Multiple Ack

Multiple acknowledgments allow a consumer to acknowledge multiple messages at once, improving efficiency when processing batches of messages.

Property Description Values
multiple Indicates whether to acknowledge multiple messages true or false

Example: Acknowledging Multiple Messages

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) => {
    // Process the message
    console.log('Received:', msg.content.toString());
    
    // Acknowledge multiple messages at once
    channel.ack(msg, true); // true = multiple acknowledgment
  }, { noAck: false });
}

consumeMessages().catch(console.error);
Note: When using multiple: true, the server acknowledges all messages up to and including the specified message. Ensure that all previous messages in the batch have been successfully processed before acknowledging.

5. Requeuing Messages

Property Description Values
requeue Controls whether a message should be requeued or discarded true or false

Example: Requeuing a Message

const amqp = require('amqplib');

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

  channel.consume(queue, (msg) => {
    if (msg && msg.content) {
      const data = JSON.parse(msg.content.toString());
      console.log('Processing:', data);

      // Simulate a processing error
      if (data.error) {
        console.log('Requeuing message...');
        channel.nack(msg, false, true); // Requeue = true
      } else {
        console.log('Message processed successfully');
        channel.ack(msg); // Acknowledge the message
      }
    }
  });
}

requeueMessage().catch(console.error);
Note: When requeue is set to true, the message is returned to the queue and can be processed again by another consumer.
Warning: Requeuing messages indefinitely can lead to infinite loops or message starvation.

6. Understanding Unacked Messages

Property Description Notes
unacked_messages Number of messages that have been delivered but not yet acknowledged Used for monitoring and debugging message flow
message_id Unique identifier for each message Used to track individual messages through the system
delivery_tag Identifier for a specific message delivery Used to acknowledge or reject specific messages

Example: Tracking Unacked Messages

const amqp = require('amqplib');

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

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

  // Set up message consumption
  channel.consume(queueName, (msg) => {
    console.log('Received message:', msg.content.toString());
    // Do not acknowledge yet
  }, { noAck: false });

  // Monitor unacked messages (pseudo-code)
  setInterval(() => {
    const unackedCount = channel.unackedMessages || 0;
    console.log(`Unacked messages: ${unackedCount}`);
  }, 5000);
}

trackUnackedMessages().catch(console.error);
Note: Unacked messages can lead to message re-delivery if the consumer disconnects or fails to acknowledge them within a specified timeout.
Warning: If unacked messages accumulate for too long, it may indicate a performance or configuration issue in your message processing pipeline.

7. Understanding Redelivered Flag

The redelivered flag is a boolean property set by the broker when a message is redelivered to a consumer. It indicates whether the message has been previously delivered and acknowledged or not.

Property Description Value
redelivered Indicates if the message was previously delivered and not acknowledged. true or false

Example: Handling Redelivered Messages

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 });

  channel.consume(queue, (msg) => {
    console.log('Redelivered:', msg.fields.redelivered);
    
    if (msg.fields.redelivered) {
      console.log('Message was previously redelivered');
    }

    // Process message
    channel.ack(msg);
  });
}

consumeMessages().catch(console.error);
Note: When a message is redelivered, it usually means that the consumer failed to acknowledge it within the specified timeout or explicitly rejected it using nack with requeue set to true.

8. Setting Ack Timeout

Property Values Description
consumer.ack.timeout 30000 Timeout in milliseconds for message acknowledgment
consumer.ack.timeout.mode auto, manual Mode of acknowledgment timeout handling

Example: Configuring Ack Timeout

{
  "consumer": {
    "ack": {
      "timeout": 30000,
      "mode": "auto"
    }
  }
}
Note: Setting a timeout ensures that unacknowledged messages are redelivered if the consumer becomes unavailable.

9. Handling Ack Failures

Example: Consumer handles nack with retry logic

const amqp = require('amqplib');

async function handleAckFailure() {
  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, async (msg) => {
    try {
      // Process message
      await processMessage(msg.content.toString());
      
      // Acknowledge successful processing
      channel.ack(msg);
    } catch (error) {
      console.error('Message processing failed:', error);
      
      // Reject and requeue for retry
      channel.nack(msg, false, true); // requeue = true
    }
  });
}

async function processMessage(data) {
  // Simulate potential failure
  if (Math.random() < 0.2) throw new Error('Random failure');
  
  console.log('Processing:', data);
}
Method Description Parameters Notes
channel.nack(msg, allUpTo, requeue) Rejects one or more messages msg: Message object
allUpTo: Boolean
requeue: Boolean		 Use requeue = true to re-queue failed messages
NEW in AMQP 0-9-1
channel.reject(msg, requeue) Rejects a single message msg: Message object
requeue: Boolean		 Legacy method, DEPRECATED in favor of nack
Note: When nack is called with requeue = true, the message returns to the queue and can be processed by another consumer.
Warning: Infinite requeue loops can occur if the same message always fails. Implement circuit breaker or dead-letter exchange logic to prevent this.

Best Practices for Ack Failures

  • Implement retry logic with exponential backoff
  • Use dead-letter exchanges (DLX) for messages that fail too many times
  • Log failed messages for debugging and monitoring
  • Ensure consumers can handle re-queued messages idempotently

10. Recovering from Nack

When a message is rejected using nack(), it can be recovered in several ways depending on the configuration and use case.

Recovery Method Description Configuration
requeue = true Requeues the message back to the queue for re-processing. nack(msg, true)
requeue = false Discards the message or sends it to a dead-letter exchange (if configured). nack(msg, false)

Example: Requeuing a Message

channel.consume(queue, (msg) => {
  if (msg.content.toString() === "error") {
    // Requeue the message for later processing
    channel.nack(msg, false); // or true to requeue
  } else {
    channel.ack(msg);
  }
});
Note: If requeue is set to false, the message will be discarded or sent to a dead-letter exchange if one is configured.

11. Using Manual Ack Best Practices

Best Practice Description Implementation
Acknowledge messages immediately after processing Ensures no duplicate processing and maintains message order 
channel.consume(queue, (msg) => {
  // Process message
  console.log("Processing: ", msg.content.toString());
  
  // Acknowledge after successful processing
  channel.ack(msg);
});
Use manual acknowledgment mode Explicitly control when messages are acknowledged 
channel.consume(queue, (msg) => {
  // Process message
  const result = processMessage(msg.content);
  
  if (result.success) {
    channel.ack(msg); // Acknowledge on success
  } else {
    channel.nack(msg, false, true); // Requeue on failure
  }
});
Handle errors gracefully Requeue messages on failure to prevent data loss 
channel.consume(queue, (msg) => {
  try {
    const data = JSON.parse(msg.content.toString());
    processBusinessLogic(data);
    
    // Acknowledge only after successful processing
    channel.ack(msg);
  } catch (error) {
    console.error("Error processing message: ", error);
    
    // Requeue the message for retry
    channel.nack(msg, false, true); // requeue = true
  }
});
Avoid long-running operations Prevent channel timeouts and reduce risk of message re-delivery 
channel.consume(queue, (msg) => {
  const processInWorker = () => {
    // Offload to worker or background task
    const result = heavyProcessing(msg.content);
    
    if (result.success) {
      channel.ack(msg);
    } else {
      channel.nack(msg, false, true);
    }
  };
  
  setImmediate(processInWorker); // Non-blocking
});
Note: Manual acknowledgments must be explicitly sent to RabbitMQ to ensure message persistence and prevent duplicate processing.

12. Monitoring Unacked Messages

Unacked messages are those that have been delivered to a consumer but not yet acknowledged. Monitoring these is crucial for ensuring message reliability and identifying potential bottlenecks or consumer issues.

Property Description Value
unacked_messages Number of messages currently unacknowledged in a queue 0
consumer_tag Identifier for the consumer that received the message consumer-1
message_id Unique identifier for the message msg-12345

Example: Using RabbitMQ Management Plugin to View Unacked Messages

curl -u guest:guest http://localhost:15672/api/queues/%2f/my-queue
Note: The management plugin provides a JSON response that includes messages_unacknowledged as part of the queue statistics.

Example: Setting Up a Consumer with Manual Acknowledgment

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 });
  
  channel.consume(queue, (msg) => {
    console.log('Received:', msg.content.toString());
    
    // Manual acknowledgment
    channel.ack(msg);
  }, { noAck: false });
}

consumeMessages();

Was this article helpful?

Related Cheat Sheets