Tuning Prefetch and Consumer Concurrency

Prefetch and concurrency are the two knobs that decide how work is distributed across a fleet, and they are the reason a queue with idle workers can still have a growing backlog. This guide applies the flow-control model from Producer-Consumer Pattern Design, part of Queue Fundamentals & Architecture.

Problem Statement

Twelve workers consume one queue. During a backlog, four of them are saturated while eight sit idle, and the p99 wait climbs into minutes. Nothing is broken: each worker prefetched fifty messages at connection time, and the four that happened to grab the long jobs are still working through their private buffers while the idle workers have nothing left to claim. You want work distributed by availability rather than by arrival order.

Prerequisites

• Measured p50 and p99 job durations per queue — every recommendation below is derived from their ratio.
• Knowledge of what each job blocks on: CPU, database, or an external API.
• Connection-pool sizes for anything the job touches, since concurrency multiplies them.
• The ability to change worker settings and restart, ideally with the drain from graceful shutdown & worker deployments.

Step 1 — Separate the Two Concepts

Prefetch (prefetch_count, worker_prefetch_multiplier, MaxNumberOfMessages) is how many messages a consumer reserves — held locally, invisible to everyone else, whether or not it is working on them. Concurrency is how many it executes at once. Prefetch above concurrency is buffering; prefetch equal to concurrency is just-in-time.

Step 2 — Choose Prefetch from Duration Spread

The right prefetch is a function of how uneven your jobs are, not of how fast your broker is.

def recommended_prefetch(p50_seconds: float, p99_seconds: float) -> int:
    """Prefetch amortises a ~1ms broker round trip. Above ~50ms per job it buys
    nothing measurable, and unevenness makes it actively harmful."""
    if p50_seconds > 0.05 or (p99_seconds / max(p50_seconds, 1e-6)) > 5:
        return 1
    if p50_seconds > 0.01:
        return 4
    return 32

recommended_prefetch(0.002, 0.003)   # 32 — tiny uniform jobs
recommended_prefetch(0.4, 12.0)      # 1  — long tail

Step 3 — Choose Concurrency from What the Job Blocks On

Concurrency should track the resource the job waits on, not the number of cores.

# CPU-bound: one process per core, minus headroom for the OS and the broker client
concurrency = max(1, os.cpu_count() - 1)

# I/O-bound: high concurrency is fine, but it is bounded by the pools it uses
concurrency = min(
    db_pool_size - 2,          # never exhaust the connection pool
    api_rate_limit_per_worker, # respect the downstream's concurrency budget
    64,                        # beyond this, context switching dominates
)

The binding constraint is almost always a pool, not the CPU. Twenty concurrent tasks against a ten-connection pool produce ten workers doing useful work and ten blocked on pool timeout — which surfaces as mysterious latency rather than as an obvious error. Size the pool first, then set concurrency below it, as in tuning the Sidekiq Redis connection pool.

Step 4 — Apply the Settings per Framework

# Celery — prefetch is a MULTIPLIER of concurrency, not an absolute count
worker_prefetch_multiplier = 1     # 1 × concurrency reserved
worker_concurrency = 8
task_acks_late = True              # required, or prefetch=1 is unsafe on crash

// BullMQ — no prefetch concept; concurrency is the only knob
new Worker('jobs', processor, { concurrency: 8, connection });

# Sidekiq — concurrency is threads; Redis pool must be at least concurrency + 2
Sidekiq.configure_server { |c| c.concurrency = 10; c.redis = { size: 12 } }

# SQS — the batch size IS the prefetch, and it multiplies the visibility need
sqs.receive_message(QueueUrl=url, MaxNumberOfMessages=1, WaitTimeSeconds=20)

Celery's multiplier semantics catch people out: worker_prefetch_multiplier = 4 with concurrency = 16 reserves sixty-four messages, not four.

Step 5 — Account for the Memory Cost

Every reserved message is held in worker memory, and with large payloads that adds up quickly.

def prefetch_memory_mb(prefetch: int, concurrency: int, avg_msg_kb: float) -> float:
    """Celery-style: total reserved = prefetch multiplier × concurrency."""
    return prefetch * concurrency * avg_msg_kb / 1024

prefetch_memory_mb(4, 16, 256)     # 16 MB — fine
prefetch_memory_mb(50, 16, 2048)   # 1600 MB — an OOM kill waiting to happen

This is another argument for the claim-check pattern: with small reference messages, prefetch stops being a memory decision at all.

Step 6 — Verify with a Distribution Test

# RabbitMQ: reserved-but-not-running is the unacked count
rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers
# jobs  84210  640  12   → 640 unacked across 12 consumers with concurrency 8
#                          means ~53 reserved each: prefetch is far too high.

Step 7 — Re-check After Every Workload Change

These settings encode assumptions about job duration, and job duration changes whenever the code does. A task that grew from 40ms to 4 seconds because it started calling a new service turns a well-tuned prefetch of 32 into a fairness bug, and nothing in the system announces that.

Make it observable rather than remembered. Export the ratio of reserved to running messages per worker as a gauge, and alert when it exceeds a few times concurrency for a sustained period — that single signal catches both a prefetch set too high and a job that quietly got slower. Pair it with a per-worker throughput panel: if the spread between the busiest and quietest worker exceeds roughly 30% while the queue is non-empty, the distribution is skewed and prefetch is the first thing to look at.

Revisit the numbers as a matter of routine when duration percentiles shift by more than a factor of two, when concurrency changes, and when a new task class is routed onto an existing queue. That last case is the most common and the least noticed: adding one long-running task to a queue of short ones changes the duration spread for every task on it, which is a good reason to route genuinely different work to genuinely different queues in the first place.

A useful habit is to record the tuned values with their justification in the same place as the queue definition — prefetch=1 because p99/p50 = 14 ages far better than a bare number, and it tells the next person what evidence would justify changing it.

Verification

1. Even distribution. Under sustained backlog, jobs-per-worker should be within roughly 20% across the fleet.
2. No idle workers with a non-empty queue. Any worker at zero utilisation while depth is positive means prefetch is hoarding.
3. Unacked ≈ concurrency. Reserved messages per consumer should be close to its concurrency, not a multiple of it.
4. Throughput did not regress. For very short jobs, confirm that reducing prefetch has not cut throughput measurably; if it has, raise it back and accept some unevenness.

Gotchas & Edge Cases

Prefetch=1 with acks_early loses work. With early acknowledgement, a crash discards the in-flight message entirely. acks_late is a prerequisite for a low prefetch to be safe.

Raising concurrency to fix a backlog often makes it worse. If the bottleneck is a downstream pool, more concurrent tasks simply queue inside the client library instead of in the broker, where you can no longer see them. Confirm where the time is spent before turning the knob.

Prefetch applies per channel, not per process. A worker opening several channels multiplies the effective reservation. Count channels before concluding the setting is low.

Long-poll batch size is prefetch on SQS. MaxNumberOfMessages=10 reserves ten and starts the visibility clock on all of them at once, so the timeout must cover the whole batch — see debugging duplicate deliveries in SQS.

Concurrency above the pool size is invisible. Blocked-on-pool tasks look like slow tasks in every dashboard. Instrument pool wait time separately or you will tune the wrong knob.

Gevent and thread pools hide CPU contention. A concurrency of 200 greenlets looks fine until one task does real computation: the event loop stalls, every other greenlet's latency spikes, and the metrics blame the queue rather than the pool. Keep CPU work out of cooperatively-scheduled pools, or drop concurrency to what one core can genuinely interleave.

Prefetch interacts with the drain window. Every reserved message is released unacknowledged at shutdown and redelivered elsewhere, so a fleet that deploys several times a day pays for a high prefetch in duplicate executions rather than in latency. That cost does not appear in any throughput benchmark, which is why prefetch tuned purely on a load test tends to be too high for production.

Autoscalers react to depth, not to distribution. A hoarding fleet with idle workers still shows a growing backlog, so the autoscaler adds workers that also idle. Fix prefetch before concluding you need more capacity.

Related

• Producer-Consumer Pattern Design — the flow-control model these settings implement.
• Backpressure strategies for fast producers — the other half of balancing arrival and service rates.
• Priority Queues & Job Fairness — why prefetch defeats broker-level priority.
• Horizontal Worker Scaling — scaling out once the distribution is actually even.