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

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

2. Celery Configuration

# 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

# 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

# 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

# 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

# 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

# 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.