Skip to main content
Mapagam

Implementing Work Queue Pattern

1. Creating Task Queue

Property Description Default Value
queue.name Name of the queue to create task_queue
durable Whether the queue should survive broker restarts true
autoDelete Whether the queue should be deleted when no consumers are connected false
exclusive Whether the queue should be exclusive to one consumer false

Example: Declaring a Task Queue in Python

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare a durable queue
channel.queue_declare(queue='task_queue', durable=True)

print(' [*] Waiting for messages in task_queue. To exit press CTRL+C')

def callback(ch, method, properties, body):
    print(f" [x] Received {body}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='task_queue', on_message_callback=callback)
channel.start_consuming()
Note: The durable parameter ensures that the queue and messages survive broker restarts.

2. Distributing Tasks

Property Description Default Value Notes
task_queue.name Name of the task queue task_queue Used to identify the queue in RabbitMQ
task_queue.durable Whether the queue survives broker restarts true Set to true for persistence
task_queue.prefetch_count Number of messages to prefetch 1 Controls load balancing between workers

Example: Publishing a Task

const amqp = require('amqplib');

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

  await channel.assertQueue(queue, { durable: true });
  const task = { message: 'Process this task' };
  channel.sendToQueue(queue, Buffer.from(JSON.stringify(task)), {
    persistent: true
  });

  console.log('Task published');
  await channel.close();
  await connection.close();
}

publishTask();

Example: Consuming a Task

const amqp = require('amqplib');

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

  await channel.assertQueue(queue, { durable: true });
  channel.prefetch(1); // Ensure fair distribution

  channel.consume(queue, async (msg) => {
    const task = JSON.parse(msg.content.toString());
    console.log('Processing:', task.message);
    
    // Simulate work
    await new Promise(resolve => setTimeout(resolve, 1000));
    
    console.log('Task completed');
    channel.ack(msg); // Acknowledge task completion
  });
}

consumeTask();
Note: Using persistent: true ensures messages survive broker restarts.
Warning: If a worker crashes while processing a task, the message will be requeued and processed by another worker.

3. Using Multiple Workers

When using multiple workers, RabbitMQ distributes messages among available consumers in a round-robin fashion. This ensures load distribution across workers.

Property Description Default Value
prefetchCount Number of messages that the broker will deliver to a consumer before acknowledging one. 1

Example: Setting up multiple workers with prefetch

const amqp = require('amqplib');

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

  await channel.assertQueue('task_queue', { durable: true });
  await channel.prefetch(1); // Process one message at a time

  channel.consume('task_queue', (msg) => {
    const body = msg.content.toString();
    console.log(`Processing: ${body}`);
    
    // Simulate work
    setTimeout(() => {
      console.log('Done');
      channel.ack(msg); // Acknowledge message
    }, 1000);
  });
}

setupWorker().catch(console.error);
Note: Using prefetch(1) ensures that messages are not delivered to another worker until the current one acknowledges the previous message. This prevents overloading a single worker.

4. Implementing Fair Dispatch

Property Description Default Value Notes
prefetchCount Number of messages to prefetch from the queue 1 Controls how many messages are delivered to a consumer at once
qos Quality of Service settings for the channel 1 Used to configure fair dispatch behavior

Example: Setting up Fair Dispatch in RabbitMQ

const amqp = require('amqplib');

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

  // Set QoS to limit prefetch count
  channel.prefetch(1);

  // Consume messages with fair dispatch
  channel.consume('work_queue', (msg) => {
    console.log("Processing message", msg.content.toString());
    channel.ack(msg);
  });
}

setupFairDispatch().catch(console.error);
Note: Setting prefetchCount to 1 ensures that only one message is delivered at a time, enabling fair distribution among consumers.

5. Ensuring Message Durability

Property Description Default
durable When set to true, makes the queue survive broker restarts. false
autoDelete When set to true, the queue is deleted when the last consumer unsubscribes. false
exclusive When set to true, the queue is only accessible by the connection that declares it. false

Example: Declaring a Durable Queue

const queueName = 'work_queue';
const durable = true;

channel.assertQueue(queueName, { durable: durable });

Example: Publishing a Durable Message

const message = 'Process this task';

channel.publish('', 'work_queue', Buffer.from(message), {
  deliveryMode: 2 // Makes the message persistent
});
Note: To ensure message durability, both the queue and the message must be configured for persistence. Messages are not persisted to disk unless explicitly marked as persistent.

6. Implementing Acknowledgments

Acknowledgments in RabbitMQ ensure that messages are not lost if a consumer crashes or becomes unavailable. By default, RabbitMQ uses autoAck set to true, which means that messages are automatically acknowledged as soon as they are delivered to the consumer. However, in many use cases, it's crucial to manually acknowledge messages to ensure reliability.

Property Description Values
autoAck Controls whether messages are automatically acknowledged true or false
deliveryTag Unique identifier for a message delivery Integer
multiple Indicates whether to acknowledge multiple messages up to the delivery tag true or false

Example: Manual Acknowledgment in Consumer

const amqp = require('amqplib');

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

  await channel.assertQueue(queue, { durable: true });
  
  // Disable auto-acknowledgment
  channel.consume(queue, (msg) => {
    console.log('Received:', msg.content.toString());
    
    // Simulate work
    setTimeout(() => {
      channel.ack(msg); // Manually acknowledge the message
    }, 1000);
  }, { noAck: false });
}

consumeMessages().catch(console.error);
Note: If autoAck is set to false, it's critical to call channel.ack() or channel.nack() to acknowledge or reject messages, respectively. Failing to do so can lead to message loss or re-delivery issues.

Example: Rejecting Messages with Nack

channel.consume(queue, (msg) => {
  try {
    // Process the message
    processMessage(msg.content);
    
    // Acknowledge successful processing
    channel.ack(msg);
  } catch (error) {
    console.error('Message processing failed:', error);
    
    // Reject the message and requeue it
    channel.nack(msg, false, true); // requeue = true
  }
});

Key Points for Acknowledgments

  • autoAck must be set to false to enable manual acknowledgments.
  • Use channel.ack() to confirm successful message processing.
  • Use channel.nack() to reject messages, optionally requeuing them.
  • Always handle errors in message processing to prevent unacknowledged messages from being lost.

7. Handling Worker Failures

When a worker fails to process a message, RabbitMQ provides mechanisms to handle the failure gracefully and ensure message reliability.

Scenario Action Description
Message Acknowledgement channel.ack(msg) Explicitly acknowledge a message after successful processing to remove it from the queue.
Nack with Requeue channel.nack(msg, true) Reject a message and requeue it for another worker to process.
Nack without Requeue channel.nack(msg, false) Reject a message and discard it or send to a dead-letter exchange.

Example: Worker Failure Handling

const amqp = require('amqplib');

async function processMessage(msg) {
  try {
    // Simulate processing
    await doWork(msg.content.toString());
    // Acknowledge successful processing
    channel.ack(msg);
  } catch (err) {
    console.error('Worker failed:', err);
    // Requeue the message for retry
    channel.nack(msg, true);
  }
}

async function doWork(data) {
  if (Math.random() < 0.2) throw new Error('Simulated failure');
  console.log('Processing:', data);
}
Note: Ensure that message processing is idempotent to avoid issues when reprocessing messages.

Failure Handling Strategies

  • Requeue: Re-deliver the message to another worker.
  • Dead Letter Exchange: Route failed messages to a DLX for logging or special handling.
  • Retry Logic: Implement retry limits to prevent infinite loops on persistent failures.

8. Using Message Priority

Property Value Description
priority 1-9 Message priority level (1 = lowest, 9 = highest)

Example: Setting Message Priority

const amqp = require('amqplib');

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

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

  const message = {
    text: 'High priority task',
    priority: 9
  };

  channel.publish('', queue, Buffer.from(JSON.stringify(message)), {
    priority: message.priority
  });

  console.log('Message sent with priority:', message.priority);
  await channel.close();
  await connection.close();
}

sendMessageWithPriority().catch(console.error);
Note: Message priority requires the priority argument to be set when publishing messages. Queues must support priority (e.g., x-max-priority queue argument).
Warning: Prioritized messages may not be processed in strict order due to queue implementation details and concurrent consumers.

9. Implementing Task Timeout

Property Value Description
task_timeout 30000 Timeout in milliseconds for task execution
retry_on_timeout false Whether to retry tasks that exceed timeout

Example: Configuring Task Timeout

const amqp = require('amqplib');

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

  const queueName = 'task_queue';
  const timeoutMs = 30000;

  await channel.assertQueue(queueName, { durable: true });
  await channel.setPrefetch(1);

  // Set timeout for task processing
  const taskHandler = (msg) => {
    setTimeout(() => {
      console.log('Task timed out');
      channel.nack(msg, false, false); // Reject and requeue
    }, timeoutMs);

    // Process the actual task here
    console.log('Processing task:', msg.content.toString());
  };

  channel.consume(queueName, taskHandler, { noAck: false });
}

configureTaskTimeout().catch(console.error);
Note: Setting a timeout ensures that long-running tasks do not block the queue indefinitely. If a task exceeds its configured timeout, it will be rejected and can optionally be requeued for retry.

10. Monitoring Queue Depth

Property Description Value
queue.length Current number of messages in the queue 0
queue.messages Total number of messages in the queue (including unacknowledged) 0
queue.consumer_count Number of active consumers for the queue 0

Example: Using RabbitMQ Management Plugin to Monitor Queue Depth

curl -u guest:guest http://localhost:15672/api/queues/%2f/my_queue
Note: The management plugin must be enabled to access queue statistics via the API.

Example: Programmatically Monitoring Queue Depth with Python

import pika
import json

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

queue_name = 'my_queue'
method_frame = channel.queue_declare(queue=queue_name, passive=True)

print(f"Queue depth: {method_frame.message_count}")

Was this article helpful?

Related Cheat Sheets