Home/Learn/RabbitMQ/Work Queues (Task Distribution)

Work Queues (Task Distribution)

Beginner

Messaging Patterns

Multiple workers consume from a single queue; with prefetch=1 and manual ack, work is evenly distributed to idle workers rather than buffered on a busy one.

Overview

The Work Queue pattern (also called the Task Queue or Competing Consumers pattern) is one of the most common and useful messaging patterns. A single queue holds tasks; multiple worker processes (consumers) compete to pick up and process those tasks. Each task is delivered to exactly one worker. This naturally distributes work across available workers and allows horizontal scaling — add more workers when throughput needs to grow. The key to fair dispatch is the combination of manual acknowledgement and prefetch=1: the broker sends the next task to a worker only after it has acknowledged the previous one, ensuring idle workers get new tasks rather than busy workers accumulating a backlog.

Work Queue Pattern — Setup and Fair Dispatch

Key components: 1. **Durable queue** — survives broker restart. 2. **Persistent messages** — survive broker restart. 3. **Manual ack** — broker only removes the task when the worker confirms completion. 4. **Prefetch=1** — the broker sends at most 1 unacked message per channel; the next is not delivered until the current one is acked.

Without prefetch=1, RabbitMQ uses round-robin: it pre-delivers 250 tasks to worker A and 250 to worker B regardless of processing speed. If worker A is slow, it accumulates 250 queued tasks while worker B sits idle.

Java — Work Queue Producer + Worker

// Producer — create work and publish
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection conn = factory.newConnection();
     Channel channel = conn.createChannel()) {

    // Durable queue — survives restart
    channel.queueDeclare("task_queue", true, false, false, null);

    String[] tasks = {"resize_image_1", "send_email_2", "process_payment_3"};
    for (String task : tasks) {
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .deliveryMode(2)  // persistent — survives restart
            .build();
        channel.basicPublish("", "task_queue", props, task.getBytes());
        System.out.println("Sent: " + task);
    }
}

// Worker — multiple instances of this can run in parallel
try (Connection conn = factory.newConnection();
     Channel channel = conn.createChannel()) {

    channel.queueDeclare("task_queue", true, false, false, null);
    channel.basicQos(1);  // fair dispatch — max 1 unacked message at a time

    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
        String task = new String(delivery.getBody());
        System.out.println("Processing: " + task);
        try {
            doWork(task);  // potentially slow
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  // ack on success
        } catch (Exception e) {
            channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); // requeue on failure
        }
    };

    channel.basicConsume("task_queue", false, deliverCallback, tag -> {});
}

Spring AMQP Work Queue

Spring AMQP's @RabbitListener + SimpleMessageListenerContainer implements the work queue pattern with minimal boilerplate. Set concurrency to run multiple consumer threads per instance, or deploy multiple pods in Kubernetes.

Java + YAML — Spring AMQP Work Queue

@Configuration
public class WorkQueueConfig {

    @Bean
    public Queue taskQueue() {
        return QueueBuilder.durable("task_queue").build();
    }
}

@Component
@RequiredArgsConstructor
public class TaskWorker {

    // Spring AMQP handles prefetch and ack automatically (MANUAL mode)
    @RabbitListener(
        queues = "task_queue",
        concurrency = "3-10"  // min 3, max 10 concurrent worker threads
    )
    public void processTask(String task, Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        try {
            log.info("Worker thread {} processing: {}", Thread.currentThread().getName(), task);
            taskService.execute(task);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("Task failed: {}", task, e);
            channel.basicNack(deliveryTag, false, false);  // → DLQ
        }
    }
}

# application.yml — set prefetch for fair dispatch
spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: MANUAL
        prefetch: 1   # essential for fair dispatch

Scaling Work Queues Horizontally

The work queue pattern scales linearly. If processing one task takes 1 second and you have 100 tasks queued, one worker takes 100 seconds, two workers take 50 seconds, ten workers take 10 seconds. In Kubernetes, scale the worker Deployment's replicas based on queue depth (RabbitMQ management API exposes queue length, which can feed KEDA for autoscaling).

YAML — KEDA Autoscaler + CLI

# Kubernetes HPA or KEDA autoscaler based on queue depth
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: task-worker-scaler
spec:
  scaleTargetRef:
    name: task-worker        # Deployment name
  minReplicaCount: 1
  maxReplicaCount: 20
  triggers:
    - type: rabbitmq
      metadata:
        host: amqp://rabbitmq:5672
        queueName: task_queue
        mode: QueueLength      # scale based on queue message count
        value: "50"            # target 50 messages per replica

# RabbitMQ Management HTTP API — queue depth for monitoring
curl -u guest:guest http://localhost:15672/api/queues/%2F/task_queue | jq '.messages'

Key Points to Remember

1Work queues distribute tasks across multiple competing workers — each task is delivered to exactly one worker.
2prefetch=1 (basicQos) + manual ack = fair dispatch: a busy worker does not receive new tasks until it acks the current one.
3Without prefetch=1, RabbitMQ round-robins messages regardless of worker speed, causing uneven load distribution.
4Durable queue + persistent messages (deliveryMode=2) ensures no task loss on broker restart.
5Scale workers horizontally by increasing Kubernetes replicas or @RabbitListener concurrency; queue depth is the scaling signal.
6KEDA can autoscale K8s Deployments based on RabbitMQ queue depth — scale to zero when queue is empty.

Interview Questions

What is a work queue pattern and when would you use it?

EasyAmazon

Why is prefetch=1 important for fair task distribution in RabbitMQ?

MediumUber

What happens to a task if a worker crashes while processing it (with manual ack)?

MediumFlipkart

How would you scale workers automatically based on queue depth in Kubernetes?

HardNetflix

A task queue has 10 000 messages and 5 workers. One worker is 10x slower than the others. How does prefetch=1 help?

MediumGoogle

Ask Aria about Work Queues (Task Distribution)

Your personal AI tutor — ask anything about this concept

Revision Status

Personal Notes

Discussion

Loading discussion…

Delayed Message Exchange

Publish-Subscribe Pattern