Background Job Processing — Celery, Sidekiq, and Bull Explained

# celery_app.py
# Celery application configuration
from celery import Celery

app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
    include=['tasks']
)

app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
    task_track_started=True,
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    task_soft_time_limit=300,
    task_time_limit=600,
)

# tasks.py
# Background task definitions for Celery
from celery_app import app
import time

@app.task(bind=True, max_retries=3, default_retry_delay=10)
def send_email(self, recipient, subject, body):
    """Send an email asynchronously with retry support."""
    try:
        print(f"Preparing email to {recipient}: {subject}")
        # Simulate SMTP call
        time.sleep(2)
        print(f"Email sent to {recipient}")
        return {"recipient": recipient, "status": "sent"}
    except ConnectionError as exc:
        raise self.retry(exc=exc, countdown=60)

@app.task(bind=True, rate_limit='10/m')
def process_image(self, image_path, output_sizes=None):
    """Process an image with rate limiting."""
    if output_sizes is None:
        output_sizes = [(800, 600), (400, 300), (150, 150)]
    results = []
    for width, height in output_sizes:
        # Simulate image processing
        time.sleep(1)
        result = {
            "input": image_path,
            "output": f"{image_path}_thumb_{width}x{height}.jpg",
            "size": f"{width}x{height}",
        }
        results.append(result)
        self.update_state(state='PROGRESS', meta={
            'current': len(results),
            'total': len(output_sizes),
        })
    print(f"Image processed: {image_path}")
    return {"results": results, "status": "completed"}

@app.task
def generate_report(user_id, report_type):
    """Generate a daily or weekly report."""
    print(f"Generating {report_type} report for user {user_id}")
    time.sleep(5)
    report_url = f"/reports/{user_id}/{report_type}/2026-06-22.pdf"
    print(f"Report generated: {report_url}")
    return {"user_id": user_id, "report_url": report_url}

// bull-queue.js
// Background job processing with Bull and Redis
const Queue = require('bull');

const emailQueue = new Queue('email', {
  redis: { host: 'localhost', port: 6379 },
  defaultJobOptions: {
    attempts: 5,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: 100,
    removeOnFail: 50,
  },
});

emailQueue.process(async (job) => {
  const { to, subject, body } = job.data;
  console.log(`[Job ${job.id}] Processing email to ${to}`);

  // Progress updates
  await job.progress(10);
  // Simulate send
  await new Promise((resolve) => setTimeout(resolve, 1500));

  await job.progress(50);
  await new Promise((resolve) => setTimeout(resolve, 1000));

  await job.progress(100);
  console.log(`[Job ${job.id}] Email sent to ${to}`);
  return { sent: true, to, timestamp: Date.now() };
});

emailQueue.on('failed', (job, err) => {
  console.error(`[Job ${job.id}] Failed attempt ${job.attemptsMade}: ${err.message}`);
});

emailQueue.on('completed', (job, result) => {
  console.log(`[Job ${job.id}] Completed: ${JSON.stringify(result)}`);
});

// Enqueue jobs
async function addEmailJobs() {
  await emailQueue.add(
    { to: 'alice@example.com', subject: 'Welcome!', body: 'Thank you for joining' },
    { priority: 1 }
  );
  await emailQueue.add(
    { to: 'bob@example.com', subject: 'Your invoice', body: 'Invoice attached' },
    { delay: 3600000 }
  );
  console.log('Jobs added to queue');
}

addEmailJobs();

# sidekiq_worker.rb
# Sidekiq background job worker
require 'sidekiq'

class ReportGenerationWorker
  include Sidekiq::Worker
  sidekiq_options queue: 'reports', retry: 3, backtrace: true

  def perform(user_id, report_type)
    logger.info "Generating #{report_type} report for user #{user_id}"

    # Simulate report generation
    sleep 4

    report_data = {
      user_id: user_id,
      type: report_type,
      generated_at: Time.now.iso8601,
      url: "/reports/#{user_id}/#{report_type}/latest"
    }

    # Store result
    Sidekiq.redis do |conn|
      conn.set("report:#{user_id}:#{report_type}", report_data.to_json)
    end

    logger.info "Report generated for user #{user_id}"
    report_data
  rescue StandardError => e
    logger.error "Report generation failed: #{e.message}"
    raise
  end
end

# Enqueue from controller
ReportGenerationWorker.perform_async(42, 'weekly')
# Schedule for later
ReportGenerationWorker.perform_in(86400, 43, 'daily')

# job_chaining.py
# Chain Celery tasks using callbacks
from celery import chain, group, chord
from tasks import send_email, process_image, generate_report

# Chain: run tasks sequentially
pipeline = chain(
    send_email.s('alice@example.com', 'Welcome!', 'Body'),
    generate_report.s(42, 'summary')
)
pipeline()

# Group: run tasks in parallel
parallel_tasks = group(
    process_image.s('/uploads/photo1.jpg'),
    process_image.s('/uploads/photo2.jpg'),
    process_image.s('/uploads/photo3.jpg'),
)
result = parallel_tasks()

# Chord: group + callback when all complete
callback = generate_report.s(42, 'summary')
workflow = chord(
    [process_image.s(f'/uploads/photo{i}.jpg') for i in range(5)],
    callback
)
workflow()

# flower_config.py
# Flower configuration for Celery monitoring
from flower.utils import template

# Custom flower template configuration
flower_config = {
    'port': 5555,
    'broker_API': 'Redis://localhost:6379/0',
    'address': '0.0.0.0',
    'db': 0,
    'max_tasks': 10000,
    'persistent': True,
    'State_save_interval': 10000,
    'auto_refresh': True,
    'natural_time': True,
    'tasks_routes': True,
    'enable_events': True,
}

# Custom metrics dashboard
# Run: Celery -A Celery_app flower --conf=flower_config.py

# API endpoints for external monitoring
# /dashboard/ -> Queue overview
# /tasks -> Active, pending, failed counts
# /workers -> Worker status and load

Celery -A Celery_app flower --port=5555 --auto_refresh=True