Worker Processes for Background Jobs
In this tutorial, you will learn about Worker Processes for Background Jobs. We cover key concepts, practical examples, and best practices to help you master this topic.
Worker processes execute background jobs asynchronously, handling concurrency, lifecycle, graceful shutdown, and scaling across multiple machines.
What You Learn
You will learn how workers pick up and execute jobs, worker concurrency models, graceful shutdown, lifecycle management, and how to run workers in production.
Why It Matters
Workers are the execution engine. A misconfigured worker crashes jobs. A worker without graceful shutdown loses tasks. Understanding worker internals helps you build reliable processing systems.
Real-World Use
DodaTech runs workers as Docker containers managed by Kubernetes. Each worker has health checks, graceful shutdown (SIGTERM), and resource limits. Workers auto-scale based on queue depth.
How Workers Work
flowchart LR
Q[Job Queue] -->|Fetch Job| W[Worker]
W -->|Process| J[JOB]
J -->|Success| A[Ack job]
J -->|Failure| R[Retry/DLQ]
W -->|Fetch Next| Q
style W fill:#f90,color:#fff
A worker continuously polls the queue, fetches jobs, executes them, and acknowledges completion. Failed jobs are retried or sent to a dead letter queue.
Simple Worker
import redis
import json
import time
import signal
import sys
r = redis.Redis()
class SimpleWorker:
def __init__(self, queues=['default']):
self.queues = queues
self.handlers = {}
self.running = True
signal.signal(signal.SIGTERM, self.stop)
signal.signal(signal.SIGINT, self.stop)
def register(self, task_name):
def decorator(func):
self.handlers[task_name] = func
return func
return decorator
def stop(self, signum=None, frame=None):
print(f"\nShutting down worker...")
self.running = False
def process_job(self, job_data):
job = json.loads(job_data)
task = job['task']
handler = self.handlers.get(task)
if not handler:
print(f"Unknown task: {task}")
return
print(f"Processing: {task}")
try:
result = handler(**job.get('data', {}))
print(f"Completed: {task} -> {result}")
return True
except Exception as e:
print(f"Failed: {task} -> {e}")
return False
def start(self):
print(f"Worker started. Queues: {self.queues}")
while self.running:
for queue in self.queues:
job_data = r.lpop(queue)
if job_data:
self.process_job(job_data)
time.sleep(0.1)
print("Worker stopped")
worker = SimpleWorker(['default', 'high'])
@worker.register('send_email')
def send_email(to, subject):
time.sleep(0.5)
return f"Sent to {to}"
@worker.register('process_file')
def process_file(path):
time.sleep(1)
return f"Processed {path}"
# Start worker
worker.start()
Expected output:
Worker started. Queues: ['default', 'high']
Worker Concurrency
import redis
import json
import time
import threading
r = redis.Redis()
class ConcurrentWorker:
def __init__(self, queue='default', concurrency=4):
self.queue = queue
self.concurrency = concurrency
self.handlers = {}
self.running = True
def register(self, name):
def decorator(func):
self.handlers[name] = func
return func
return decorator
def worker_loop(self, worker_id):
print(f" Worker {worker_id} started")
while self.running:
job_data = r.brpop(self.queue, timeout=2)
if job_data:
_, data = job_data
job = json.loads(data)
handler = self.handlers.get(job['task'])
if handler:
print(f"[W{worker_id}] Processing: {job['task']}")
handler(**job.get('data', {}))
print(f"[W{worker_id}] Done: {job['task']}")
def start(self):
print(f"Starting {self.concurrency} workers for queue '{self.queue}'")
threads = []
for i in range(self.concurrency):
t = threading.Thread(target=self.worker_loop, args=(i,), daemon=True)
t.start()
threads.append(t)
for t in threads:
t.join()
worker = ConcurrentWorker('process', concurrency=3)
@worker.register('slow_job')
def slow_job(data):
time.sleep(1)
print(f" Completed: {data}")
# Queues jobs
for i in range(6):
r.lpush('process', json.dumps({'task': 'slow_job', 'data': f'item_{i}'}))
worker.start()
Expected output:
Starting 3 workers for queue 'process'
Worker 0 started
Worker 1 started
Worker 2 started
[W0] Processing: slow_job
[W1] Processing: slow_job
[W2] Processing: slow_job
Completed: item_0
Completed: item_1
Completed: item_2
[W0] Done: slow_job
[W1] Done: slow_job
[W2] Done: slow_job
Graceful Shutdown
import signal
import time
import redis
import json
r = redis.Redis()
class GracefulWorker:
def __init__(self):
self.running = True
self.current_job = None
signal.signal(signal.SIGTERM, self._handle_signal)
signal.signal(signal.SIGINT, self._handle_signal)
def _handle_signal(self, signum, frame):
print(f"\nReceived signal {signum}, finishing current job...")
self.running = False
def start(self):
print("Worker starting...")
while self.running:
job_data = r.brpop('default', timeout=2)
if job_data and self.running:
_, data = job_data
job = json.loads(data)
self.current_job = job
print(f"Processing: {job['task']}")
time.sleep(3)
print(f"Completed: {job['task']}")
self.current_job = None
print("Worker stopped gracefully")
# Test: send SIGTERM while worker is busy
worker = GracefulWorker()
# worker.start() # Run and press Ctrl+C
Worker Health Check
import redis
import time
import json
r = redis.Redis()
class HealthCheckWorker:
def __init__(self, name='worker1'):
self.name = name
self.heartbeat_key = f"worker:heartbeat:{name}"
def send_heartbeat(self):
r.setex(self.heartbeat_key, 10, json.dumps({
'name': self.name,
'time': time.time(),
'status': 'alive',
}))
def start(self):
print(f"Worker {self.name} started")
while True:
self.send_heartbeat()
time.sleep(5)
def check_workers():
workers = r.keys('worker:heartbeat:*')
for key in workers:
data = json.loads(r.get(key))
age = time.time() - data['time']
status = 'ALIVE' if age < 15 else 'DEAD'
print(f" {data['name']}: {status} (last: {age:.0f}s ago)")
# Start worker in thread
import threading
w = HealthCheckWorker('worker-a')
t = threading.Thread(target=w.start, daemon=True)
t.start()
time.sleep(3)
check_workers()
Expected output:
Worker worker-a started
worker-a: ALIVE (last: 2s ago)
Common Mistakes
1. Not Acknowledging Jobs
If a worker crashes after processing but before acknowledging, the job is reprocessed. Use explicit acknowledgements to prevent both data loss and duplicate processing.
2. Blocking Worker on Slow Jobs
A worker processing a slow job cannot fetch new jobs. Use concurrent workers: one worker per CPU core for CPU-bound tasks, or many workers for I/O-bound tasks.
3. Not Handling Worker Crashes
A crashed worker leaves in-progress jobs stuck. Use heartbeats to detect dead workers and timeouts to redeliver stuck jobs.
4. Ignoring Graceful Shutdown
Killing a worker with SIGKILL loses the current job. Use SIGTERM, complete the current job, then exit. Implement signal handlers.
5. Overloading Workers
More concurrent workers than CPU cores cause context switching overhead. For CPU-bound tasks, worker count should match CPU cores. For I/O-bound tasks, higher concurrency is fine.
Practice Questions
1. What does a worker do?
A worker continuously polls the job queue, fetches jobs, executes the corresponding handler, and acknowledges completion.
2. How does worker concurrency work?
Multiple worker processes or threads execute jobs in parallel. Each worker runs independently, fetching and processing its own jobs.
3. What is graceful shutdown?
The worker finishes its current job before exiting. It catches SIGTERM, completes the running job, acknowledges it, then stops.
4. How do you handle a crashed worker?
Use heartbeats to detect worker death. Set job timeouts so that unacknowledged jobs are redelivered to other workers after a timeout.
Challenge
Design a worker deployment for a video processing platform: CPU-intensive transcoding (1 worker per core), I/O-intensive upload/download (gevent-style, 50 concurrent), health check (ping every 5s), graceful shutdown (max 30s for current job), and auto-scaling (add workers when queue > 100).
FAQ
Mini Project: Worker Pool
import redis
import json
import time
import threading
import signal
r = redis.Redis()
class WorkerPool:
def __init__(self, queue, num_workers=4):
self.queue = queue
self.num_workers = num_workers
self.handlers = {}
self.workers = []
self.running = True
def task(self, name):
def decorator(func):
self.handlers[name] = func
return func
return decorator
def run_worker(self, wid):
while self.running:
try:
job_data = r.brpop(self.queue, timeout=3)
if job_data and self.running:
_, data = job_data
job = json.loads(data)
handler = self.handlers.get(job['task'])
if handler:
print(f"[W{wid}] Start: {job['task']}")
handler(**job.get('data', {}))
print(f"[W{wid}] End: {job['task']}")
except Exception as e:
print(f"[W{wid}] Error: {e}")
def start(self):
print(f"Starting {self.num_workers} workers")
for i in range(self.num_workers):
t = threading.Thread(target=self.run_worker, args=(i,), daemon=True)
t.start()
self.workers.append(t)
def stop(self):
self.running = False
print("Stopping workers...")
pool = WorkerPool('work', num_workers=3)
@pool.task('email')
def email_task(to, subject):
time.sleep(0.5)
print(f" Email sent to {to}")
@pool.task('report')
def report_task(name):
time.sleep(1)
print(f" Report: {name} generated")
pool.start()
for i in range(6):
task = 'email' if i % 2 == 0 else 'report'
data = {'to': f'user{i}@example.com', 'subject': 'Hello'} if task == 'email' else {'name': f'report_{i}'}
r.lpush('work', json.dumps({'task': task, 'data': data}))
time.sleep(5)
pool.stop()
Expected output:
Starting 3 workers
[W0] Start: email
[W1] Start: report
[W2] Start: email
Email sent to user0@example.com
Report: report_1 generated
Email sent to user2@example.com
[W0] End: email
[W1] End: report
[W2] End: email
What's Next
Now that you understand workers, explore Bull Queue for Node.js for JavaScript background jobs, then learn about Sidekiq for Ruby.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro