# Background Workers: Process Heavy Tasks

# Background Workers: Process Heavy Tasks with Separate Worker Processes

## Problem

Web applications often face performance bottlenecks when handling CPU-intensive or time-consuming operations synchronously. When users trigger heavy tasks—like image processing, PDF generation, data analysis, or sending bulk emails—the main application thread blocks, causing:

- **Slow response times** for all users
- **Request timeouts** on long-running operations
- **Poor user experience** with frozen interfaces
- **Resource contention** between concurrent requests
- **Inability to scale** horizontally

Traditional synchronous processing makes applications unresponsive and unreliable.

## Solution

Implement a **background worker architecture** that separates heavy computational tasks from the main request-response cycle. This involves:

1. **Task Queue**: A message broker (Redis, RabbitMQ) stores pending tasks
2. **Worker Processes**: Dedicated processes consume and execute tasks asynchronously
3. **Job Scheduler**: Manages task distribution and retry logic
4. **Status Tracking**: Monitor task progress and results
5. **Error Handling**: Graceful failure recovery and logging

This decoupling allows the main application to respond immediately while workers process tasks in the background, improving responsiveness and scalability.

## Code Implementation

### 1. Basic Setup with Celery and Redis

```python
# requirements.txt
celery==5.3.1
redis==5.0.0
flask==3.0.0
pillow==10.0.0
```

### 2. Celery Configuration

```python
# celery_config.py
from celery import Celery
from kombu import Exchange, Queue
import os

app = Celery('myapp')

# Redis as message broker
app.conf.broker_url = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
app.conf.result_backend = os.getenv('REDIS_URL', 'redis://localhost:6379/0')

# Task configuration
app.conf.task_serializer = 'json'
app.conf.accept_content = ['json']
app.conf.result_serializer = 'json'
app.conf.timezone = 'UTC'
app.conf.enable_utc = True

# Task routing and queues
app.conf.task_default_queue = 'default'
app.conf.task_queues = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('high_priority', Exchange('high_priority'), routing_key='high_priority'),
    Queue('low_priority', Exchange('low_priority'), routing_key='low_priority'),
)

# Task routing rules
app.conf.task_routes = {
    'tasks.image_processing.*': {'queue': 'high_priority'},
    'tasks.email.*': {'queue': 'default'},
    'tasks.analytics.*': {'queue': 'low_priority'},
}

# Retry configuration
app.conf.task_acks_late = True
app.conf.worker_prefetch_multiplier = 1
app.conf.task_max_retries = 3
```

### 3. Define Background Tasks

```python
# tasks.py
from celery_config import app
from celery import Task
import time
import logging
from PIL import Image
import io
import requests

logger = logging.getLogger(__name__)

class CallbackTask(Task):
    """Task with error callbacks"""
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.error(f'Task {task_id} failed: {exc}')
    
    def on_success(self, result, task_id, args, kwargs):
        logger.info(f'Task {task_id} succeeded')

@app.task(base=CallbackTask, bind=True, max_retries=3)
def process_image(self, image_url, width, height):
    """
    Download and resize image asynchronously
    """
    try:
        self.update_state(state='PROGRESS', meta={'current': 0, 'total': 100})
        
        # Download image
        response = requests.get(image_url, timeout=30)
        response.raise_for_status()
        
        self.update_state(state='PROGRESS', meta={'current': 50, 'total': 100})
        
        # Process image
        img = Image.open(io.BytesIO(response.content))
        img.thumbnail((width, height), Image.Resampling.LANCZOS)
        
        # Save result
        output = io.BytesIO()
        img.save(output, format='JPEG')
        output.seek(0)
        
        self.update_state(state='PROGRESS', meta={'current': 100, 'total': 100})
        
        return {
            'status': 'success',
            'size': len(output.getvalue()),
            'dimensions': img.size
        }
    
    except Exception as exc:
        logger.exception(f'Image processing failed: {exc}')
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

@app.task(bind=True)
def send_bulk_emails(self, recipient_list, subject, template):
    """
    Send emails to multiple recipients
    """
    total = len(recipient_list)
    
    for idx, recipient in enumerate(recipient_list):
        try:
            # Simulate email sending
            time.sleep(0.5)
            logger.info(f'Email sent to {recipient}')
            
            # Update progress
            self.update_state(
                state='PROGRESS',
                meta={'current': idx + 1, 'total': total}
            )
        except Exception as e:
            logger.error(f'Failed to send email to {recipient}: {e}')
    
    return {'sent': total, 'failed': 0}

@app.task
def generate_report(data_source, report_type):
    """
    Generate complex reports
    """
    logger.info(f'Generating {report_type} report from {data_source}')
    
    # Simulate heavy computation
    time.sleep(5)
    
    return {
        'report_id': 'RPT-12345',
        'type': report_type,
        'rows': 1000,
        'generated_at': time.time()
    }

@app.task
def cleanup_old_files(days=30):
    """
    Periodic task to clean up old files
    """
    logger.info(f'Cleaning up files older than {days} days')
    # Implementation here
    return {'deleted': 42}
```

### 4. Flask Application Integration

```python
# app.py
from flask import Flask, request, jsonify
from celery_config import app as celery_app
from tasks import process_image, send_bulk_emails, generate_report
import logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.route('/api/process-image', methods=['POST'])
def submit_image_task():
    """
    Submit image processing task
    """
    data = request.json
    image_url = data.get('image_url')
    width = data.get('width', 800)
    height = data.get('height', 600)
    
    if not image_url:
        return {'error': 'image_url required'}, 400
    
    # Submit task to queue
    task = process_image.apply_async(
        args=[image_url, width, height],
        queue='high_priority',
        priority=9
    )
    
    return {
        'task_id': task.id,
        'status': 'queued',
        'status_url': f'/api/task-status/{task.id}'
    }, 202

@app.route('/api/send-emails', methods=['POST'])
def submit_email_task():
    """
    Submit bulk email task
    """
    data = request.json
    recipients = data.get('recipients', [])
    subject = data.get('subject', 'Hello')
    template = data.get('template', 'default')
    
    if not recipients:
        return {'error': 'recipients required'}, 400
    
    task = send_bulk_emails.apply_async(
        args=[recipients, subject, template],
        queue='default'
    )
    
    return {
        'task_id': task.id,
        'recipients_count': len(recipients),
        'status_url': f'/api/task-status/{task.id}'
    }, 202

@app.route('/api/generate-report', methods=['POST'])
def submit_report_task():
    """
    Submit report generation task
    """
    data = request.json
    data_source = data.get('data_source')
    report_type = data.get('report_type', 'summary')
    
    task = generate_report.apply_async(
        args=[data_source, report_type],
        queue='low_priority'
    )
    
    return {
        'task_id': task.id,
        'status_url': f'/api/task-status/{task.id}'
    }, 202

@app.route('/api/task-status/<task_id>', methods=['GET'])
def get_task_status(task_id):
    """
    Get task status and progress
    """
    task = celery_app.AsyncResult(task_id)
    
    response = {
        'task_id': task_id,
        'status': task.status,
    }
    
    if task.status == 'PENDING':
        response['progress'] = 0
    elif task.status == 'PROGRESS':
        response['progress'] = task.info.get('current', 0)
        response['total'] = task.info.get('total', 100)
    elif task.status == 'SUCCESS':
        response['result'] = task.result
        response['progress'] = 100
    elif task.status == 'FAILURE':
        response['error'] = str(task.info)
        response['traceback'] = task.traceback
    
    return response

@app.route('/api/task-cancel/<task_id>', methods=['POST'])
def cancel_task(task_id):
    """
    Cancel a running task
    """
    task = celery_app.AsyncResult(task_id)
    task.revoke(terminate=True)
    
    return {'task_id': task_id, 'status': 'cancelled'}

@app.route('/health', methods=['GET'])
def health_check():
    """
    Health check endpoint
    """
    try:
        # Check if Celery is responsive
        celery_app.control.inspect().active()
        return {'status': 'healthy'}, 200
    except Exception as e:
        logger.error(f'Health check failed: {e}')
        return {'status': 'unhealthy', 'error': str(e)}, 503

if __name__ == '__main__':
    app.run(debug=True, port=5000)
```

### 5. Worker Process Startup

```bash
# Start worker with multiple concurrency options

# Single worker with 4 processes
celery -A tasks worker --loglevel=info --concurrency=4

# Worker with gevent (for I/O-bound tasks)
celery -A tasks worker --pool=gevent --concurrency=1000 --loglevel=info

# Multiple workers for different queues
celery -A tasks worker -Q high_priority --concurrency=8 --loglevel=info
celery -A tasks worker -Q default --concurrency=4 --loglevel=info
celery -A tasks worker -Q low_priority --concurrency=2 --loglevel=info

# Worker with autoscaling
celery -A tasks worker --autoscale=10,3 --loglevel=info
```

### 6. Monitoring and Management

```python
# monitor.py
from celery_config import app
from celery.app.control import Inspect
import json

def get_worker_stats():
    """Get statistics from all workers"""
    inspect = Inspect(app=app)
    
    stats = {
        'active_tasks': inspect.active(),
        'registered_tasks': inspect.registered(),
        'stats': inspect.stats(),
    }
    
    return stats

def get_queue_length():
    """Get pending tasks in each queue"""
    inspect = Inspect(app=app)
    reserved = inspect.reserved()
    
    return reserved

def restart_worker(worker_name):
    """Restart a specific worker"""
    app.control.shutdown(destination=[worker_name])

def purge_queue(queue_name):
    """Clear all tasks from a queue"""
    app.control.purge()

if __name__ == '__main__':
    print(json.dumps(get_worker_stats(), indent=2))
```

### 7. Docker Compose Setup

```yaml
# docker-compose.yml
version: '3.8'

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  app:
    build: .
    ports:
      - "5000:5000"
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    command: python app.py

  worker_high:
    build: .
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    command: celery -A tasks worker -Q high_priority --concurrency=8

  worker_default:
    build: .
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    command: celery -A tasks worker -Q default --concurrency=4

  worker_low:
    build: .
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    command: celery -A tasks worker -Q low_priority --concurrency=2

volumes:
  redis_data:
```

## Tips & Best Practices

### 1. **Task Design**
- Keep tasks **idempotent** (safe to retry)
- Use **small, focused tasks** rather than monolithic ones
- Avoid passing large objects; use IDs instead
- Set appropriate **timeouts** to prevent hanging

### 2. **Error Handling**
- Implement **exponential backoff** for retries
- Log failures comprehensively
- Use **dead letter queues** for permanently failed tasks
- Monitor retry rates

### 3. **Performance Optimization**
- Use **task routing** to separate workloads
- Implement **autoscaling** based on queue depth
- Choose appropriate **concurrency models** (processes vs. gevent)
- Monitor **worker utilization** and adjust accordingly

### 4. **Monitoring & Observability**
- Use **Flower** for real-time monitoring
- Track **task metrics** (duration, success rate)
- Set up **alerts** for worker failures
- Log all task executions

### 5. **Scaling Strategies**
- Deploy **multiple workers** across machines
- Use **priority queues** for critical tasks
- Implement **rate limiting** to prevent overload
- Consider **task batching** for bulk operations

### 6. **Production Considerations**
- Run workers with **process managers** (systemd, supervisor)
- Use **persistent message brokers** (RabbitMQ for reliability)
- Implement **graceful shutdown** handling
- Set up **health checks** and auto-recovery

Background workers transform application architecture, enabling responsive systems that handle heavy workloads reliably and scale efficiently.