RabbitMQ Consumer

You are a message queue engineer. The user wants to build a reliable RabbitMQ consumer that handles acknowledgments correctly and uses prefetch to control message flow.

What to check first

• Verify RabbitMQ server is running: rabbitmq-diagnostics -s ping
• Check that amqplib is installed: npm list amqplib (or pip list | grep pika for Python)
• Confirm the queue exists or will be declared by the consumer

Steps

1. Connect to RabbitMQ using amqplib.connect() with proper error handling for reconnection
2. Create a channel with connection.createChannel() to isolate this consumer's operations
3. Declare the queue using channel.assertQueue() with durable: true to persist it across restarts
4. Set prefetch with channel.prefetch(1) to pull only one message at a time (prevents overwhelming the worker)
5. Consume messages using channel.consume() with noAck: false to enable manual acknowledgment
6. Process the message in the callback and call channel.ack(msg) only after successful processing
7. Implement error handling to call channel.nack(msg, false, true) to requeue failed messages
8. Add graceful shutdown to close the channel and connection on process termination

Code

const amqp = require('amqplib');

const RABBITMQ_URL = process.env.RABBITMQ_URL || 'amqp://guest:guest@localhost';
const QUEUE_NAME = 'task_queue';
const PREFETCH_COUNT = 1;

async function startConsumer() {
  let connection;
  let channel;

  try {
    // Step 1: Connect to RabbitMQ
    connection = await amqp.connect(RABBITMQ_URL);
    console.log('Connected to RabbitMQ');

    // Step 2: Create a channel
    channel = await connection.createChannel();

    // Step 3: Declare the queue (idempotent)
    await channel.assertQueue(QUEUE_NAME, {
      durable: true,
      arguments: {
        'x-message-ttl': 86400000, // 24 hours
      },
    });

    // Step 4: Set prefetch to 1 (process one message at a time)
    await channel.prefetch(PREFETCH_COUNT);

    console.log(`Waiting for messages in queue: ${QUEUE_NAME}`);

    // Step 5: Start consuming with manual acknowledgment
    channel.consume(QUEUE_NAME, async (msg) => {
      if (!msg) return;

      try {
        // Parse the message content
        const content = JSON.parse(msg.content.toString());
        console.log(`Received message:`, content);

        // Simulate async work
        await processMessage(content);

        // Step

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change