Worker Scalability and Throughput — Complete Guide
In this tutorial, you will learn about Worker Scalability and Throughput. We cover key concepts, practical examples, and best practices to help you master this topic.
Scale background job workers horizontally and vertically, optimize throughput with prefetch limits, batching, and concurrency tuning for peak performance.
What You Learn
You will learn how to scale workers horizontally (more instances) and vertically (more threads per instance), optimize throughput with prefetch settings and batching, measure worker performance, and avoid common scaling pitfalls.
Why It Matters
As your application grows, job volume increases. A single worker processing one job at a time cannot keep up. Understanding scalability ensures your background job system handles peak load without backpressure, queue growth, or resource exhaustion.
Real-World Use
DodaTech processes 50,000 file scans per hour during peak times. The worker pool autoscales from 10 to 100 workers across 20 machines. Each worker processes 5 jobs concurrently with prefetch limits tuned to avoid overwhelming the database.
Horizontal Scaling
import redis
import json
import time
import threading
r = redis.Redis()
class HorizontallyScalableWorker:
def __init__(self, worker_id, queue='jobs'):
self.worker_id = worker_id
self.queue = queue
self.running = True
self.jobs_processed = 0
def process(self, job):
time.sleep(0.1)
self.jobs_processed += 1
return True
def work(self):
while self.running:
job_data = r.brpop(self.queue, timeout=1)
if job_data:
_, data = job_data
job = json.loads(data)
self.process(job)
if self.jobs_processed % 10 == 0:
print(f"Worker {self.worker_id}: {self.jobs_processed} jobs done")
def stop(self):
self.running = False
def scale_out(n):
workers = []
for i in range(n):
w = HorizontallyScalableWorker(f'H-{i}')
t = threading.Thread(target=w.work, daemon=True)
t.start()
workers.append(w)
return workers
for i in range(50):
r.lpush('jobs', json.dumps({'id': f'bulk-{i}', 'task': 'process'}))
workers = scale_out(5)
time.sleep(3)
total = sum(w.jobs_processed for w in workers)
print(f"Total processed by 5 workers: {total}")
for w in workers:
w.stop()
Expected output:
Worker H-0: 10 jobs done
Worker H-2: 10 jobs done
Worker H-1: 10 jobs done
Worker H-3: 10 jobs done
Worker H-4: 10 jobs done
Total processed by 5 workers: 50
Vertical Scaling with Concurrency
import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor
class ConcurrentWorker:
def __init__(self, worker_id, concurrency=4):
self.worker_id = worker_id
self.concurrency = concurrency
self.executor = ThreadPoolExecutor(max_workers=concurrency)
self.running = True
def handle_job(self, job):
time.sleep(random.uniform(0.05, 0.2))
print(f" [{self.worker_id}] Processed: {job['id']}")
return job['id']
def process_batch(self, jobs):
futures = [self.executor.submit(self.handle_job, j) for j in jobs]
results = [f.result() for f in futures]
return results
def stop(self):
self.running = False
self.executor.shutdown(wait=True)
worker = ConcurrentWorker('V1', concurrency=4)
jobs = [{'id': f'job-{i}'} for i in range(8)]
start = time.time()
results = worker.process_batch(jobs)
elapsed = time.time() - start
print(f"Processed {len(results)} jobs in {elapsed:.2f}s with concurrency {worker.concurrency}")
worker.stop()
Expected output:
[V1] Processed: job-0
[V1] Processed: job-2
[V1] Processed: job-1
[V1] Processed: job-3
[V1] Processed: job-4
[V1] Processed: job-5
[V1] Processed: job-6
[V1] Processed: job-7
Processed 8 jobs in ~0.25s with concurrency 4
Prefetch Limits
import redis
import json
import time
r = redis.Redis()
class PrefetchWorker:
def __init__(self, worker_id, prefetch_count=5):
self.worker_id = worker_id
self.prefetch_count = prefetch_count
def fetch_batch(self):
pipeline = r.pipeline()
for _ in range(self.prefetch_count):
pipeline.rpop(self.queue)
results = pipeline.execute()
return [json.loads(r) for r in results if r]
def process_batch(self, jobs):
for job in jobs:
print(f" [{self.worker_id}] {job['task']}: {job['id']}")
time.sleep(0.05)
def work(self, queue='jobs'):
self.queue = queue
while True:
batch = self.fetch_batch()
if not batch:
time.sleep(0.5)
continue
self.process_batch(batch)
for i in range(10):
r.lpush('jobs', json.dumps({'id': f'j-{i}', 'task': 'scan'}))
worker = PrefetchWorker('P1', prefetch_count=5)
import threading
t = threading.Thread(target=worker.work, daemon=True)
t.start()
time.sleep(1)
Expected output:
[P1] scan: j-9
[P1] scan: j-8
[P1] scan: j-7
[P1] scan: j-6
[P1] scan: j-5
Throughput Measurement
import time
import threading
import redis
import json
r = redis.Redis()
class ThroughputMonitor:
def __init__(self):
self.metrics_key = 'worker:metrics'
def start_monitoring(self):
def loop():
while True:
time.sleep(5)
jobs_processed = r.get('counter:processed')
queue_depth = r.llen('jobs')
rate = int(jobs_processed or 0) / 5
print(f"Throughput: {rate:.1f} jobs/sec, Queue: {queue_depth}, Total: {int(jobs_processed or 0)}")
thread = threading.Thread(target=loop, daemon=True)
thread.start()
def record_job(self):
r.incr('counter:processed')
monitor = ThroughputMonitor()
monitor.start_monitoring()
for i in range(20):
r.lpush('jobs', json.dumps({'id': f'metric-{i}'}))
time.sleep(2)
for _ in range(15):
data = r.rpop('jobs')
if data:
monitor.record_job()
time.sleep(0.1)
time.sleep(4)
r.delete('counter:processed', 'jobs')
Expected output:
Throughput: 3.0 jobs/sec, Queue: 5, Total: 15
Batching for Throughput
import time
class BatchProcessor:
def __init__(self, batch_size=10, flush_interval=5):
self.batch_size = batch_size
self.flush_interval = flush_interval
self.buffer = []
self.last_flush = time.time()
def add(self, item):
self.buffer.append(item)
if len(self.buffer) >= self.batch_size:
self.flush()
def flush(self):
if not self.buffer:
return
batch = self.buffer[:]
self.buffer = []
print(f"Flushing batch of {len(batch)} items")
time.sleep(0.2)
self.last_flush = time.time()
return batch
def flush_interval_check(self):
if time.time() - self.last_flush >= self.flush_interval and self.buffer:
self.flush()
def shutdown(self):
self.flush()
processor = BatchProcessor(batch_size=5, flush_interval=10)
for i in range(12):
processor.add(f"item-{i}")
time.sleep(0.05)
processor.shutdown()
Expected output:
Flushing batch of 5 items
Flushing batch of 5 items
Flushing batch of 2 items
Common Mistakes
1. Overloading the Database
Scaling workers without throttling database connections overwhelms the database. Use connection pooling and worker concurrency limits.
2. Ignoring Prefetch Limits
High prefetch counts grab too many jobs from the queue, causing starvation for other workers. Set prefetch to 1-5 for balanced distribution.
3. No Backpressure
When the queue grows indefinitely, workers fall behind and memory usage increases. Implement backpressure: stop accepting new jobs when the queue exceeds a threshold.
4. Blind Concurrency Increase
More threads per worker does not always mean more throughput. CPU-bound jobs need fewer threads than I/O-bound jobs. Profile before tuning.
5. No Monitoring
Scaling blindly without metrics leads to over-provisioning or under-provisioning. Track queue depth, processing time, and worker utilization.
Practice Questions
1. What is the difference between horizontal and vertical scaling?
Horizontal scaling adds more worker instances. Vertical scaling adds more threads or resources to existing workers.
2. What is prefetch count?
The number of jobs a worker fetches from the queue at once. Higher prefetch improves throughput but may cause uneven distribution.
3. How does concurrency affect throughput?
More concurrent threads Process more jobs in parallel, but only up to the point where resources (CPU, database connections, I/O) become saturated.
4. What is backpressure?
A mechanism that slows or stops accepting new jobs when the system is overloaded, preventing unbounded queue growth.
Challenge
Build a scalable worker system for processing 10,000 emails per hour. Use 4 worker machines with 8 threads each, prefetch count of 5, batch database writes of 50 records, implement backpressure when the queue exceeds 5,000 items, and report throughput metrics every 10 seconds.
FAQ
Mini Project: Scalable Worker Pool
import redis
import json
import time
import threading
from concurrent.futures import ThreadPoolExecutor
r = redis.Redis()
class ScalableWorkerPool:
def __init__(self, pool_name, queue, min_workers=2, max_workers=20, concurrency=4):
self.pool_name = pool_name
self.queue = queue
self.min_workers = min_workers
self.max_workers = max_workers
self.concurrency = concurrency
self.workers = {}
self.executor = ThreadPoolExecutor(max_workers=concurrency)
def get_queue_depth(self):
return r.llen(self.queue)
def worker_loop(self, worker_id):
while True:
job_data = r.brpop(self.queue, timeout=2)
if job_data:
_, data = job_data
job = json.loads(data)
self.executor.submit(self.process_job, worker_id, job)
def process_job(self, worker_id, job):
time.sleep(0.05)
return True
def scale(self):
depth = self.get_queue_depth()
current = len(self.workers)
target = current
if depth > 20 and current < self.max_workers:
target = min(depth // 5, self.max_workers)
elif depth < 5 and current > self.min_workers:
target = max(self.min_workers, current - 1)
while len(self.workers) < target:
wid = f"{self.pool_name}-w{len(self.workers) + 1}"
t = threading.Thread(target=self.worker_loop, args=(wid,), daemon=True)
t.start()
self.workers[wid] = t
print(f"Scaled up: {wid} (target={target})")
return current, target
def shutdown(self):
self.executor.shutdown(wait=True)
pool = ScalableWorkerPool('email', 'email_queue', min_workers=1, max_workers=5, concurrency=4)
for i in range(30):
r.lpush('email_queue', json.dumps({'id': f'email-{i}', 'to': f'user{i}@x.com'}))
before, after = pool.scale()
print(f"Scaled from {before} to {after} workers")
time.sleep(3)
depth = pool.get_queue_depth()
print(f"Queue depth after processing: {depth}")
pool.shutdown()
r.delete('email_queue')
Expected output:
Scaled up: email-w1 (target=5)
Scaled up: email-w2 (target=5)
Scaled up: email-w3 (target=5)
Scaled up: email-w4 (target=5)
Scaled up: email-w5 (target=5)
Scaled from 0 to 5 workers
Queue depth after processing: 0
What's Next
Now that you understand worker scalability, build the background jobs mini project to apply everything you have learned about queues, workers, scheduling, retries, deduplication, dependencies, distributed processing, and scalability.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro