celery

Celery：分布式任务队列

概述

Celery 是一个适用于 Python 的分布式任务队列系统，可在多个工作进程间异步执行后台任务。它支持任务调度、重试和工作流，并可与 Django、Flask 和 FastAPI 无缝集成。

适用场景

• 后台处理：将耗时操作交由后台执行（邮件、文件处理、报告）
• 定时任务：类似 Cron 的周期性任务（清理、备份、数据同步）
• 分布式计算：在多个工作进程/服务器间处理任务
• 异步工作流：串联、分组并编排复杂的任务依赖关系
• 实时处理：处理 Webhook、通知和数据管道
• 负载均衡：将 CPU 密集型工作分配给多个工作进程

不适用场景：

• 简单的异步 I/O（改用 asyncio）
• 实时请求/响应（改用异步 Web 框架）
• 要求亚秒级延迟（改用内存队列）
• 基础设施需求极少（改用 RQ 或 Huey 等更简单的替代方案）

快速开始

安装

# Basic installation
pip install celery

# With Redis broker
pip install celery[redis]

# With RabbitMQ broker
pip install celery[amqp]

# Full batteries (recommended)
pip install celery[redis,msgpack,auth,cassandra,elasticsearch,s3,sqs]

基本设置

# celery_app.py
from celery import Celery

# Create Celery app with Redis broker
app = Celery(
    'myapp',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1'
)

# Configuration
app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
)

# Define a task
@app.task
def add(x, y):
    return x + y

@app.task
def send_email(to, subject, body):
    # Simulate email sending
    import time
    time.sleep(2)
    print(f"Email sent to {to}: {subject}")
    return {"status": "sent", "to": to}

运行工作进程

# Start worker
celery -A celery_app worker --loglevel=info

# Multiple workers with concurrency
celery -A celery_app worker --concurrency=4 --loglevel=info

# Named worker for specific queues
celery -A celery_app worker -Q emails,reports --loglevel=info

执行任务

# Call task asynchronously
result = add.delay(4, 6)

# Wait for result
print(result.get(timeout=10))  # 10

# Apply async with options
result = send_email.apply_async(
    args=['[email protected]', 'Hello', 'Welcome!'],
    countdown=60  # Execute after 60 seconds
)

# Check task state
print(result.status)  # PENDING, STARTED, SUCCESS, FAILURE

---

核心概念

架构组件

Broker：存储任务的消息队列

• Redis（适用于大多数场景，推荐）
• RabbitMQ（企业级，较复杂）
• Amazon SQS（无服务器，原生支持 AWS）

工作进程：执行任务的进程

• 从 Broker 获取任务
• 执行任务代码
• 将结果存储在后端

结果后端：用于存储任务结果

• Redis（快速，内存中存储）
• 数据库（PostgreSQL、MySQL）
• S3（用于存储大型结果）
• Cassandra、Elasticsearch（适用于特定场景）

Beat 调度器：周期性任务调度器

• 类似 Cron 的调度
• 基于间隔的任务
• 将调度信息存储在数据库或文件中

任务状态

PENDING → STARTED → SUCCESS
                 → RETRY → SUCCESS
                 → FAILURE

• PENDING：任务正在队列中等待
• STARTED：工作进程已获取任务
• SUCCESS：任务已成功完成
• FAILURE：任务引发异常
• RETRY：任务将在失败后重试
• REVOKED：任务在执行前已取消

---

Broker 设置

Redis 配置

# celery_config.py
broker_url = 'redis://localhost:6379/0'
result_backend = 'redis://localhost:6379/1'

# With authentication
broker_url = 'redis://:password@localhost:6379/0'

# Redis Sentinel (high availability)
broker_url = 'sentinel://localhost:26379;sentinel://localhost:26380'
broker_transport_options = {
    'master_name': 'mymaster',
    'sentinel_kwargs': {'password': 'password'},
}

# Redis connection pool settings
broker_pool_limit = 10
broker_connection_retry = True
broker_connection_retry_on_startup = True
broker_connection_max_retries = 10

RabbitMQ 配置

# Basic RabbitMQ
broker_url = 'amqp://guest:guest@localhost:5672//'

# With virtual host
broker_url = 'amqp://user:password@localhost:5672/myvhost'

# High availability (multiple brokers)
broker_url = [
    'amqp://user:password@host1:5672//',
    'amqp://user:password@host2:5672//',
]

# RabbitMQ-specific settings
broker_heartbeat = 30
broker_pool_limit = 10

Amazon SQS 配置

# AWS SQS (serverless)
broker_url = 'sqs://'
broker_transport_options = {
    'region': 'us-east-1',
    'queue_name_prefix': 'myapp-',
    'visibility_timeout': 3600,
    'polling_interval': 1,
}

# With custom credentials
import boto3
broker_transport_options = {
    'region': 'us-east-1',
    'predefined_queues': {
        'default': {
            'url': 'https://sqs.us-east-1.amazonaws.com/123456789/myapp-default',
        }
    }
}

---

任务基础

任务定义

from celery import Task, shared_task
from celery_app import app

# Method 1: Decorator
@app.task
def simple_task(x, y):
    return x + y

# Method 2: Shared task (framework-agnostic)
@shared_task
def framework_task(data):
    return process(data)

# Method 3: Task class (advanced)
class CustomTask(Task):
    def on_success(self, retval, task_id, args, kwargs):
        print(f"Task {task_id} succeeded with {retval}")

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print(f"Task {task_id} failed: {exc}")

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        print(f"Task {task_id} retrying: {exc}")

@app.task(base=CustomTask)
def monitored_task(x):
    return x * 2

任务选项

@app.task(
    name='custom.task.name',           # Custom task name
    bind=True,                          # Bind task instance as first arg
    ignore_result=True,                 # Don't store result (performance)
    max_retries=3,                      # Max retry attempts
    default_retry_delay=60,             # Retry delay in seconds
    rate_limit='100/h',                 # Rate limiting
    time_limit=300,                     # Hard time limit (kills task)
    soft_time_limit=240,                # Soft time limit (raises exception)
    serializer='json',                  # Task serializer
    compression='gzip',                 # Compress large messages
    priority=5,                         # Task priority (0-9)
    queue='high_priority',              # Target queue
    routing_key='priority.high',        # Routing key
    acks_late=True,                     # Acknowledge after execution
    reject_on_worker_lost=True,         # Reject if worker dies
)
def advanced_task(self, data):
    try:
        return process(data)
    except Exception as exc:
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

任务上下文（bind=True）

@app.task(bind=True)
def context_aware_task(self, x, y):
    # Access task metadata
    print(f"Task ID: {self.request.id}")
    print(f"Task Name: {self.name}")
    print(f"Args: {self.request.args}")
    print(f"Kwargs: {self.request.kwargs}")
    print(f"Retries: {self.request.retries}")
    print(f"Delivery Info: {self.request.delivery_info}")

    # Manual retry
    try:
        result = risky_operation(x, y)
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60, max_retries=3)

    return result

---

任务执行

Delay 与 Apply Async

# delay() - Simple async execution
result = add.delay(4, 6)

# apply_async() - Full control
result = add.apply_async(
    args=(4, 6),
    kwargs={'extra': 'data'},

    # Timing options
    countdown=60,                    # Execute after N seconds
    eta=datetime(2025, 12, 1, 10, 0),  # Execute at specific time
    expires=3600,                    # Task expires after N seconds

    # Routing options
    queue='math',                    # Target specific queue
    routing_key='math.add',          # Custom routing key
    exchange='tasks',                # Target exchange
    priority=9,                      # High priority

    # Execution options
    serializer='json',               # Message serializer
    compression='gzip',              # Compress payload
    retry=True,                      # Auto-retry on failure
    retry_policy={
        'max_retries': 3,
        'interval_start': 0,
        'interval_step': 0.2,
        'interval_max': 0.2,
    },

    # Task linking
    link=log_result.s(),             # Success callback
    link_error=handle_error.s(),     # Error callback
)

# Check result
if result.ready():
    print(result.get())  # Get result (blocks)
    print(result.result)  # Get result (non-blocking)

任务签名

from celery import signature

# Create signature (doesn't execute)
sig = add.signature((2, 2), countdown=10)
sig = add.s(2, 2)  # Shorthand

# Partial arguments (currying)
partial = add.s(2)  # One arg fixed
result = partial.apply_async(args=(4,))  # Add second arg

# Immutable signature (args can't be replaced)
immutable = add.si(2, 2)

# Clone and modify
new_sig = sig.clone(countdown=60)

# Execute signature
result = sig.delay()
result = sig.apply_async()
result = sig()  # Synchronous execution

结果处理

# Basic result retrieval
result = add.delay(4, 6)
value = result.get(timeout=10)  # Blocks until complete

# Non-blocking result check
if result.ready():
    print(result.result)

# Result states
print(result.status)      # PENDING, STARTED, SUCCESS, FAILURE
print(result.successful())  # True if SUCCESS
print(result.failed())      # True if FAILURE

# Result metadata
print(result.traceback)   # Exception traceback if failed
print(result.info)        # Task return value or exception

# Forget result (free memory)
result.forget()

# Revoke task (cancel)
result.revoke(terminate=True)  # Kill running task
add.AsyncResult(task_id).revoke()  # Revoke by ID

---

任务路由

队列配置

# Define queues
from kombu import Queue, Exchange

app.conf.task_queues = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('high_priority', Exchange('priority'), routing_key='priority.high'),
    Queue('low_priority', Exchange('priority'), routing_key='priority.low'),
    Queue('emails', Exchange('tasks'), routing_key='tasks.email'),
    Queue('reports', Exchange('tasks'), routing_key='tasks.report'),
)

# Default queue
app.conf.task_default_queue = 'default'
app.conf.task_default_exchange = 'tasks'
app.conf.task_default_routing_key = 'default'

任务路由规则

# Route specific tasks to queues
app.conf.task_routes = {
    'myapp.tasks.send_email': {'queue': 'emails'},
    'myapp.tasks.generate_report': {'queue': 'reports', 'priority': 9},
    'myapp.tasks.*': {'queue': 'default'},
}

# Function-based routing
def route_task(name, args, kwargs, options, task=None, **kw):
    if 'email' in name:
        return {'queue': 'emails', 'routing_key': 'email.send'}
    elif 'report' in name:
        return {'queue': 'reports', 'priority': 5}
    return {'queue': 'default'}

app.conf.task_routes = (route_task,)

工作进程队列绑定

# Worker consuming specific queues
celery -A myapp worker -Q emails,reports --loglevel=info

# Multiple workers for different queues
celery -A myapp worker -Q high_priority -c 4 --loglevel=info
celery -A myapp worker -Q default -c 2 --loglevel=info
celery -A myapp worker -Q low_priority -c 1 --loglevel=info

优先级队列

# Configure priority support
app.conf.task_queue_max_priority = 10
app.conf.task_default_priority = 5

# Send task with priority
high_priority_task.apply_async(args=(), priority=9)
low_priority_task.apply_async(args=(), priority=1)

# Priority-based routing
app.conf.task_routes = {
    'critical_task': {'queue': 'default', 'priority': 10},
    'background_task': {'queue': 'default', 'priority': 1},
}

---

周期性任务

Celery Beat 设置

# celery_config.py
from celery.schedules import crontab, solar

# Periodic task schedule
beat_schedule = {
    # Run every 30 seconds
    'add-every-30-seconds': {
        'task': 'myapp.tasks.add',
        'schedule': 30.0,
        'args': (16, 16)
    },

    # Run every morning at 7:30 AM
    'send-daily-report': {
        'task': 'myapp.tasks.send_daily_report',
        'schedule': crontab(hour=7, minute=30),
    },

    # Run every Monday morning
    'weekly-cleanup': {
        'task': 'myapp.tasks.cleanup',
        'schedule': crontab(hour=0, minute=0, day_of_week=1),
    },

    # Run on specific days
    'monthly-report': {
        'task': 'myapp.tasks.monthly_report',
        'schedule': crontab(hour=0, minute=0, day_of_month='1'),
        'kwargs': {'month_offset': 1}
    },

    # Solar schedule (sunrise/sunset)
    'wake-up-at-sunrise': {
        'task': 'myapp.tasks.morning_routine',
        'schedule': solar('sunrise', -37.81, 144.96),  # Melbourne
    },
}

app.conf.beat_schedule = beat_schedule

Crontab 模式

from celery.schedules import crontab

# Every minute
crontab()

# Every 15 minutes
crontab(minute='*/15')

# Every hour at :30
crontab(minute=30)

# Every day at midnight
crontab(hour=0, minute=0)

# Every weekday at 5 PM
crontab(hour=17, minute=0, day_of_week='1-5')

# Every Monday, Wednesday, Friday at noon
crontab(hour=12, minute=0, day_of_week='mon,wed,fri')

# First day of month
crontab(hour=0, minute=0, day_of_month='1')

# Last day of month (use day_of_month='28-31' with logic in task)
crontab(hour=0, minute=0, day_of_month='28-31')

# Quarterly (every 3 months)
crontab(hour=0, minute=0, day_of_month='1', month_of_year='*/3')

运行 Beat 调度器

# Start beat scheduler
celery -A myapp beat --loglevel=info

# Beat with custom scheduler
celery -A myapp beat --scheduler django_celery_beat.schedulers:DatabaseScheduler

# Combine worker and beat (development only)
celery -A myapp worker --beat --loglevel=info

动态调度（django-celery-beat）

pip install django-celery-beat

# settings.py (Django)
INSTALLED_APPS = [
    'django_celery_beat',
]

# Migrate database
python manage.py migrate django_celery_beat

# Run beat with database scheduler
celery -A myapp beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

# Create periodic task via Django admin or ORM
from django_celery_beat.models import PeriodicTask, IntervalSchedule, CrontabSchedule
import json

# Create interval schedule (every 10 seconds)
schedule, created = IntervalSchedule.objects.get_or_create(
    every=10,
    period=IntervalSchedule.SECONDS,
)

PeriodicTask.objects.create(
    interval=schedule,
    name='Import feed every 10 seconds',
    task='myapp.tasks.import_feed',
    args=json.dumps(['https://example.com/feed']),
)

# Create crontab schedule
schedule, created = CrontabSchedule.objects.get_or_create(
    minute='0',
    hour='*/4',  # Every 4 hours
    day_of_week='*',
    day_of_month='*',
    month_of_year='*',
)

PeriodicTask.objects.create(
    crontab=schedule,
    name='Hourly cleanup',
    task='myapp.tasks.cleanup',
)

---

工作流（Canvas）

链

from celery import chain

# Sequential execution
result = chain(add.s(2, 2), add.s(4), add.s(8))()
# Equivalent to: add(add(add(2, 2), 4), 8)
# Result: 16

# Shorthand syntax
result = (add.s(2, 2) | add.s(4) | add.s(8))()

# Chain with different tasks
workflow = (
    fetch_data.s(url) |
    process_data.s() |
    save_results.s()
)
result = workflow.apply_async()

组

from celery import group

# Parallel execution
job = group([
    add.s(2, 2),
    add.s(4, 4),
    add.s(8, 8),
])
result = job.apply_async()

# Wait for all results
results = result.get(timeout=10)  # [4, 8, 16]

# Group with callbacks
job = group([
    process_item.s(item) for item in items
]) | summarize_results.s()

Chord

from celery import chord

# Group with callback
job = chord([
    fetch_url.s(url) for url in urls
])(combine_results.s())

# Example: Process multiple files, then merge
workflow = chord([
    process_file.s(file) for file in files
])(merge_results.s())

result = workflow.apply_async()

Map 和 Starmap

from celery import group

# Map: Apply same task to list of args
results = add.map([(2, 2), (4, 4), (8, 8)])

# Starmap: Unpack arguments
results = add.starmap([(2, 2), (4, 4), (8, 8)])

# Equivalent to:
results = group([add.s(2, 2), add.s(4, 4), add.s(8, 8)])()

复杂工作流

from celery import chain, group, chord

# Parallel processing with sequential steps
workflow = chain(
    # Step 1: Fetch data
    fetch_data.s(source),

    # Step 2: Process in parallel
    group([
        process_chunk.s(chunk_id) for chunk_id in range(10)
    ]),

    # Step 3: Aggregate results
    aggregate.s(),

    # Step 4: Save to database
    save_results.s()
)

# Nested chords
workflow = chord([
    chord([
        subtask.s(item) for item in chunk
    ])(process_chunk.s())
    for chunk in chunks
])(final_callback.s())

# Real-world example: Report generation
generate_report = chain(
    fetch_user_data.s(user_id),
    chord([
        calculate_stats.s(),
        fetch_transactions.s(),
        fetch_activity.s(),
    ])(combine_sections.s()),
    render_pdf.s(),
    send_email.s(user_email)
)

---

错误处理

自动重试

@app.task(
    autoretry_for=(RequestException, IOError),  # Auto-retry these exceptions
    retry_kwargs={'max_retries': 5},             # Max 5 retries
    retry_backoff=True,                          # Exponential backoff
    retry_backoff_max=600,                       # Max 10 minutes backoff
    retry_jitter=True,                           # Add randomness to backoff
)
def fetch_url(url):
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

手动重试

@app.task(bind=True, max_retries=3)
def process_data(self, data):
    try:
        result = external_api_call(data)
        return result
    except TemporaryError as exc:
        # Retry after 60 seconds
        raise self.retry(exc=exc, countdown=60)
    except PermanentError as exc:
        # Don't retry, log and fail
        logger.error(f"Permanent error: {exc}")
        raise
    except Exception as exc:
        # Exponential backoff
        raise self.retry(
            exc=exc,
            countdown=2 ** self.request.retries,
            max_retries=3
        )

错误回调

@app.task
def on_error(request, exc, traceback):
    """Called when task fails"""
    logger.error(f"Task {request.id} failed: {exc}")
    send_alert(f"Task failure: {request.task}", str(exc))

@app.task
def risky_task(data):
    return process(data)

# Link error callback
risky_task.apply_async(
    args=(data,),
    link_error=on_error.s()
)

任务失败处理

from celery import Task

class CallbackTask(Task):
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        """Handle task failure"""
        logger.error(f"Task {task_id} failed with {exc}")
        # Send notification
        send_notification('Task Failed', str(exc))

    def on_success(self, retval, task_id, args, kwargs):
        """Handle task success"""
        logger.info(f"Task {task_id} succeeded: {retval}")

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        """Handle task retry"""
        logger.warning(f"Task {task_id} retrying: {exc}")

@app.task(base=CallbackTask)
def monitored_task(x):
    if x < 0:
        raise ValueError("Negative value")
    return x * 2

异常处理模式

@app.task(bind=True)
def robust_task(self, data):
    # Categorize exceptions
    try:
        return process(data)

    except NetworkError as exc:
        # Transient error - retry
        raise self.retry(exc=exc, countdown=60, max_retries=5)

    except ValidationError as exc:
        # Permanent error - don't retry
        logger.error(f"Invalid data: {exc}")
        return {'status': 'failed', 'error': str(exc)}

    except DatabaseError as exc:
        # Critical error - retry with exponential backoff
        backoff = min(2 ** self.request.retries * 60, 3600)
        raise self.retry(exc=exc, countdown=backoff, max_retries=10)

    except Exception as exc:
        # Unknown error - retry limited times
        if self.request.retries < 3:
            raise self.retry(exc=exc, countdown=120)
        else:
            # Max retries exceeded - fail and alert
            logger.critical(f"Task failed after retries: {exc}")
            send_alert('Critical Task Failure', str(exc))
            raise

---

监控与管理

任务事件

# Enable events
app.conf.worker_send_task_events = True
app.conf.task_send_sent_event = True

# Event listeners
from celery import signals

@signals.task_prerun.connect
def task_prerun_handler(sender=None, task_id=None, task=None, args=None, kwargs=None, **extra):
    print(f"Task {task.name}[{task_id}] starting")

@signals.task_postrun.connect
def task_postrun_handler(sender=None, task_id=None, task=None, retval=None, **extra):
    print(f"Task {task.name}[{task_id}] completed: {retval}")

@signals.task_failure.connect
def task_failure_handler(sender=None, task_id=None, exception=None, traceback=None, **extra):
    print(f"Task {task_id} failed: {exception}")

@signals.task_retry.connect
def task_retry_handler(sender=None, task_id=None, reason=None, **extra):
    print(f"Task {task_id} retrying: {reason}")

Flower 监控

# Install Flower
pip install flower

# Start Flower
celery -A myapp flower --port=5555

# Access dashboard
# http://localhost:5555

# Flower configuration
flower_basic_auth = ['admin:password']
flower_persistent = True
flower_db = 'flower.db'
flower_max_tasks = 10000

检查工作进程

from celery_app import app

# Get active tasks
i = app.control.inspect()
print(i.active())

# Get scheduled tasks
print(i.scheduled())

# Get reserved tasks
print(i.reserved())

# Get worker stats
print(i.stats())

# Get registered tasks
print(i.registered())

# Revoke task
app.control.revoke(task_id, terminate=True)

# Shutdown worker
app.control.shutdown()

# Pool restart
app.control.pool_restart()

# Rate limit
app.control.rate_limit('myapp.tasks.slow_task', '10/m')

命令行检查

# List active tasks
celery -A myapp inspect active

# List scheduled tasks
celery -A myapp inspect scheduled

# Worker stats
celery -A myapp inspect stats

# Registered tasks
celery -A myapp inspect registered

# Revoke task
celery -A myapp control revoke <task_id>

# Shutdown workers
celery -A myapp control shutdown

# Purge all tasks
celery -A myapp purge

自定义指标

@app.task(bind=True)
def tracked_task(self, data):
    from prometheus_client import Counter, Histogram

    task_counter = Counter('celery_tasks_total', 'Total tasks')
    task_duration = Histogram('celery_task_duration_seconds', 'Task duration')

    with task_duration.time():
        result = process(data)
        task_counter.inc()
        return result

---

框架集成

Django 集成

# myproject/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# myproject/__init__.py
from .celery import app as celery_app
__all__ = ('celery_app',)

# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TIMEZONE = 'UTC'
CELERY_ENABLE_UTC = True

# Task in Django app
# myapp/tasks.py
from celery import shared_task
from django.core.mail import send_mail

@shared_task
def send_email_task(subject, message, recipient):
    send_mail(subject, message, '[email protected]', [recipient])
    return f"Email sent to {recipient}"

# Use in views
from myapp.tasks import send_email_task

def my_view(request):
    send_email_task.delay('Hello', 'Welcome!', '[email protected]')
    return HttpResponse('Email queued')

FastAPI 集成

# celery_app.py
from celery import Celery

celery_app = Celery(
    'fastapi_app',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1'
)

@celery_app.task
def process_data(data: dict):
    # Long-running task
    import time
    time.sleep(10)
    return {"processed": data, "status": "complete"}

# main.py
from fastapi import FastAPI, BackgroundTasks
from celery_app import process_data

app = FastAPI()

@app.post("/process")
async def process_endpoint(data: dict):
    # Option 1: FastAPI BackgroundTasks (simple, in-process)
    # background_tasks.add_task(process_data, data)

    # Option 2: Celery (distributed, persistent)
    task = process_data.delay(data)
    return {"task_id": task.id, "status": "queued"}

@app.get("/status/{task_id}")
async def check_status(task_id: str):
    from celery.result import AsyncResult
    task = AsyncResult(task_id, app=celery_app)

    return {
        "task_id": task_id,
        "status": task.status,
        "result": task.result if task.ready() else None
    }

何时使用 Celery，何时使用 FastAPI BackgroundTasks：

• FastAPI BackgroundTasks：简单的即发即弃任务（日志记录、清理）
• Celery：分布式处理、重试、调度和任务结果

Flask 集成

# celery_app.py
from celery import Celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config['CELERY_BROKER_URL'],
        backend=app.config['CELERY_RESULT_BACKEND']
    )
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery

# app.py
from flask import Flask
from celery_app import make_celery

app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/1'

celery = make_celery(app)

@celery.task
def send_email(to, subject, body):
    with app.app_context():
        # Use Flask-Mail or similar
        mail.send(Message(subject, recipients=[to], body=body))

@app.route('/send')
def send_route():
    send_email.delay('[email protected]', 'Hello', 'Welcome!')
    return 'Email queued'

---

测试策略

急切模式（同步执行）

# conftest.py (pytest)
import pytest
from celery_app import app

@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'memory://',
        'result_backend': 'cache+memory://',
        'task_always_eager': True,          # Execute tasks synchronously
        'task_eager_propagates': True,      # Propagate exceptions
    }

# Test tasks
def test_add_task():
    result = add.delay(4, 6)
    assert result.get() == 10

def test_task_failure():
    with pytest.raises(ValueError):
        failing_task.delay()

使用真实 Broker 进行测试

# conftest.py
import pytest
from celery_app import app

@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'redis://localhost:6379/15',  # Test database
        'result_backend': 'redis://localhost:6379/15',
    }

@pytest.fixture
def celery_worker(celery_app):
    """Start worker for tests"""
    with celery_app.Worker() as worker:
        yield worker

def test_async_task(celery_worker):
    result = async_task.delay(data)
    assert result.get(timeout=10) == expected

模拟外部依赖

from unittest.mock import patch, MagicMock

@app.task
def fetch_and_process(url):
    response = requests.get(url)
    return process(response.json())

def test_fetch_and_process():
    with patch('requests.get') as mock_get:
        mock_get.return_value.json.return_value = {'data': 'test'}

        result = fetch_and_process.delay('http://example.com')
        assert result.get() == expected_result
        mock_get.assert_called_once_with('http://example.com')

测试周期性任务

from celery.schedules import crontab

def test_periodic_task_schedule():
    from celery_app import app

    schedule = app.conf.beat_schedule['daily-report']
    assert schedule['task'] == 'myapp.tasks.daily_report'
    assert schedule['schedule'] == crontab(hour=0, minute=0)

def test_periodic_task_execution():
    # Test task logic directly
    result = daily_report()
    assert result['status'] == 'complete'

集成测试

import pytest
from celery_app import app

@pytest.fixture(scope='module')
def celery_app():
    app.conf.update(
        broker_url='redis://localhost:6379/15',
        result_backend='redis://localhost:6379/15',
    )
    return app

@pytest.fixture(scope='module')
def celery_worker(celery_app):
    with celery_app.Worker() as worker:
        yield worker

def test_workflow(celery_worker):
    from celery import chain

    workflow = chain(
        fetch_data.s(url),
        process_data.s(),
        save_results.s()
    )

    result = workflow.apply_async()
    output = result.get(timeout=30)

    assert output['status'] == 'saved'

---

生产环境模式

工作进程配置

# Production worker with autoscaling
celery -A myapp worker \
  --autoscale=10,3 \
  --max-tasks-per-child=1000 \
  --time-limit=300 \
  --soft-time-limit=240 \
  --loglevel=info \
  --logfile=/var/log/celery/worker.log \
  --pidfile=/var/run/celery/worker.pid

# Multiple specialized workers
celery multi start \
  worker1 -A myapp -Q high_priority -c 4 --max-tasks-per-child=100 \
  worker2 -A myapp -Q default -c 2 --max-tasks-per-child=1000 \
  worker3 -A myapp -Q low_priority -c 1 --autoscale=3,1

# Graceful shutdown
celery multi stop worker1 worker2 worker3
celery multi stopwait worker1 worker2 worker3  # Wait for tasks to finish

配置最佳实践

# production_config.py
import os

# Broker settings
broker_url = os.getenv('CELERY_BROKER_URL', 'redis://localhost:6379/0')
broker_connection_retry_on_startup = True
broker_pool_limit = 50

# Result backend
result_backend = os.getenv('CELERY_RESULT_BACKEND', 'redis://localhost:6379/1')
result_expires = 3600  # 1 hour

# Serialization
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'UTC'
enable_utc = True

# Performance
worker_prefetch_multiplier = 4  # Tasks to prefetch per worker
worker_max_tasks_per_child = 1000  # Restart worker after N tasks (prevent memory leaks)
task_acks_late = True  # Acknowledge after task completes
task_reject_on_worker_lost = True  # Requeue if worker dies

# Reliability
task_track_started = True  # Track when task starts
task_time_limit = 300  # 5 minutes hard limit
task_soft_time_limit = 240  # 4 minutes soft limit

# Logging
worker_log_format = '[%(asctime)s: %(levelname)s/%(processName)s] %(message)s'
worker_task_log_format = '[%(asctime)s: %(levelname)s/%(processName)s][%(task_name)s(%(task_id)s)] %(message)s'

Systemd 服务

# /etc/systemd/system/celery.service
[Unit]
Description=Celery Service
After=network.target redis.target

[Service]
Type=forking
User=celery
Group=celery
WorkingDirectory=/opt/myapp
Environment="PATH=/opt/myapp/venv/bin"
ExecStart=/opt/myapp/venv/bin/celery multi start worker1 \
    -A myapp \
    --pidfile=/var/run/celery/%n.pid \
    --logfile=/var/log/celery/%n%I.log \
    --loglevel=INFO
ExecStop=/opt/myapp/venv/bin/celery multi stopwait worker1 \
    --pidfile=/var/run/celery/%n.pid
ExecReload=/opt/myapp/venv/bin/celery multi restart worker1 \
    -A myapp \
    --pidfile=/var/run/celery/%n.pid \
    --logfile=/var/log/celery/%n%I.log \
    --loglevel=INFO
Restart=always

[Install]
WantedBy=multi-user.target

# /etc/systemd/system/celerybeat.service
[Unit]
Description=Celery Beat Service
After=network.target redis.target

[Service]
Type=simple
User=celery
Group=celery
WorkingDirectory=/opt/myapp
Environment="PATH=/opt/myapp/venv/bin"
ExecStart=/opt/myapp/venv/bin/celery -A myapp beat \
    --loglevel=INFO \
    --pidfile=/var/run/celery/beat.pid
Restart=always

[Install]
WantedBy=multi-user.target

Sentry 集成

# Install
pip install sentry-sdk

# Configuration
import sentry_sdk
from sentry_sdk.integrations.celery import CeleryIntegration

sentry_sdk.init(
    dsn="https://your-sentry-dsn",
    integrations=[CeleryIntegration()],
    traces_sample_rate=0.1,  # 10% of transactions
)

# Tasks are automatically tracked
@app.task
def my_task(x):
    # Exceptions automatically sent to Sentry
    return risky_operation(x)

速率限制

# Global rate limiting
app.conf.task_default_rate_limit = '100/m'  # 100 tasks per minute

# Per-task rate limiting
@app.task(rate_limit='10/m')
def rate_limited_task(x):
    return expensive_operation(x)

# Dynamic rate limiting
app.control.rate_limit('myapp.tasks.slow_task', '5/m')

# Token bucket rate limiting
@app.task(rate_limit='10/s')
def api_call(endpoint):
    return requests.get(endpoint)

健康检查

# health.py
from celery_app import app

def check_celery_health():
    """Health check endpoint"""
    try:
        # Ping workers
        i = app.control.inspect()
        stats = i.stats()

        if not stats:
            return {'status': 'unhealthy', 'reason': 'No workers available'}

        # Check broker connection
        result = app.control.ping(timeout=1.0)
        if not result:
            return {'status': 'unhealthy', 'reason': 'Workers not responding'}

        return {'status': 'healthy', 'workers': len(stats)}
    except Exception as e:
        return {'status': 'unhealthy', 'error': str(e)}

# FastAPI health endpoint
@app.get("/health/celery")
async def celery_health():
    return check_celery_health()

---

性能优化

任务优化

# Use ignore_result for fire-and-forget tasks
@app.task(ignore_result=True)
def send_notification(user_id, message):
    # Don't need result, save backend overhead
    notify(user_id, message)

# Compression for large payloads
@app.task(compression='gzip')
def process_large_data(data):
    return analyze(data)

# Serialization choice
@app.task(serializer='msgpack')  # Faster than JSON
def fast_task(data):
    return process(data)

工作进程调优

# Worker concurrency
worker_concurrency = 4  # CPU-bound: num_cores
worker_concurrency = 20  # I/O-bound: higher value

# Prefetch multiplier (how many tasks to prefetch)
worker_prefetch_multiplier = 4  # Balance: 4x concurrency

# Task acknowledgment
task_acks_late = True  # Acknowledge after completion (reliability)
task_acks_late = False  # Acknowledge on receipt (performance)

# Memory management
worker_max_tasks_per_child = 1000  # Restart worker after N tasks
worker_max_memory_per_child = 200000  # Restart after 200MB

数据库结果后端优化

# Use Redis instead of database for results
result_backend = 'redis://localhost:6379/1'

# If using database, optimize
result_backend = 'db+postgresql://user:pass@localhost/celery'
database_engine_options = {
    'pool_size': 20,
    'pool_recycle': 3600,
}

# Reduce result expiry time
result_expires = 3600  # 1 hour instead of default 24 hours

任务分块

from celery import group

# Bad: One task per item (overhead)
for item in large_list:
    process_item.delay(item)

# Good: Chunk items
def chunks(lst, n):
    for i in range(0, len(lst), n):
        yield lst[i:i + n]

@app.task
def process_batch(items):
    return [process_item(item) for item in items]

# Process in batches of 100
job = group(process_batch.s(chunk) for chunk in chunks(large_list, 100))
result = job.apply_async()

连接池

# Redis connection pool
broker_pool_limit = 50  # Max connections to broker
redis_max_connections = 50

# Database connection pool
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    'postgresql://user:pass@localhost/db',
    poolclass=QueuePool,
    pool_size=20,
    max_overflow=0,
)

---

常见用例

发送邮件

@app.task(bind=True, max_retries=3)
def send_email_task(self, to, subject, body, attachments=None):
    try:
        msg = EmailMessage(subject, body, '[email protected]', [to])
        if attachments:
            for filename, content, mimetype in attachments:
                msg.attach(filename, content, mimetype)
        msg.send()
        return {'status': 'sent', 'to': to}
    except SMTPException as exc:
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

# Bulk email with rate limiting
@app.task(rate_limit='100/m')
def send_bulk_email(recipients, subject, template):
    for recipient in recipients:
        send_email_task.delay(recipient, subject, render_template(template, recipient))

生成报告

@app.task(bind=True, time_limit=600)
def generate_report(self, report_type, user_id, start_date, end_date):
    # Update progress
    self.update_state(state='PROGRESS', meta={'current': 0, 'total': 100})

    # Fetch data
    data = fetch_report_data(report_type, start_date, end_date)
    self.update_state(state='PROGRESS', meta={'current': 30, 'total': 100})

    # Generate PDF
    pdf = render_pdf(data)
    self.update_state(state='PROGRESS', meta={'current': 70, 'total': 100})

    # Upload to S3
    url = upload_to_s3(pdf, f'reports/{user_id}/{report_type}.pdf')
    self.update_state(state='PROGRESS', meta={'current': 90, 'total': 100})

    # Send notification
    send_email_task.delay(
        get_user_email(user_id),
        'Report Ready',
        f'Your report is ready: {url}'
    )

    return {'status': 'complete', 'url': url}

# Check progress
from celery.result import AsyncResult

task = AsyncResult(task_id)
if task.state == 'PROGRESS':
    print(task.info)  # {'current': 30, 'total': 100}

数据处理管道

from celery import chain, group

@app.task
def fetch_data(source):
    return download(source)

@app.task
def clean_data(raw_data):
    return clean(raw_data)

@app.task
def transform_data(clean_data):
    return transform(clean_data)

@app.task
def load_data(transformed_data):
    save_to_database(transformed_data)
    return {'status': 'loaded', 'rows': len(transformed_data)}

# ETL pipeline
etl_pipeline = chain(
    fetch_data.s('https://api.example.com/data'),
    clean_data.s(),
    transform_data.s(),
    load_data.s()
)

result = etl_pipeline.apply_async()

Webhook 处理

@app.task(bind=True, autoretry_for=(RequestException,), max_retries=5)
def process_webhook(self, webhook_data):
    # Validate signature
    if not verify_signature(webhook_data):
        raise ValueError("Invalid signature")

    # Process event
    event_type = webhook_data['type']

    if event_type == 'payment.success':
        update_order_status(webhook_data['order_id'], 'paid')
        send_confirmation_email.delay(webhook_data['customer_email'])

    elif event_type == 'payment.failed':
        notify_admin.delay('Payment Failed', webhook_data)

    return {'status': 'processed', 'event': event_type}

# FastAPI webhook endpoint
@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
    data = await request.json()
    process_webhook.delay(data)
    return {"status": "queued"}

图像处理

from celery import group, chord

@app.task
def resize_image(image_path, size):
    from PIL import Image
    img = Image.open(image_path)
    img.thumbnail(size)
    output_path = f"{image_path}_{size[0]}x{size[1]}.jpg"
    img.save(output_path)
    return output_path

@app.task
def upload_to_cdn(image_paths):
    urls = []
    for path in image_paths:
        url = cdn_upload(path)
        urls.append(url)
    return urls

# Generate multiple sizes and upload
def process_uploaded_image(image_path):
    sizes = [(800, 600), (400, 300), (200, 150), (100, 100)]

    workflow = chord([
        resize_image.s(image_path, size) for size in sizes
    ])(upload_to_cdn.s())

    return workflow.apply_async()

---

替代方案对比

Celery 与 RQ（Redis Queue）对比

RQ：仅依赖 Redis 的简易任务队列

适用 RQ 的场景：

• 简单用例（无需路由，基本重试）
• 仅使用 Redis 的基础设施
• 仅支持 Python 3
• 规模较小（<1000 个任务/分钟）

适用 Celery 的场景：

• 复杂工作流（链、弦）
• 多种代理选项
• 高级路由和优先级
• 大规模（>1000 个任务/分钟）
• 定期任务

# RQ Example
from redis import Redis
from rq import Queue

redis_conn = Redis()
q = Queue(connection=redis_conn)

job = q.enqueue(my_function, arg1, arg2)
result = job.result

Celery 与 Huey 对比

Huey：依赖项极少的轻量级任务队列

适用 Huey 的场景：

• 中小型项目
• 配置要求极少
• 仅使用 Redis 或内存
• 简单的定期任务

适用 Celery 的场景：

• 企业级应用
• 复杂的任务依赖关系
• 多种代理/后端选项
• 高级监控需求

# Huey Example
from huey import RedisHuey

huey = RedisHuey('myapp')

@huey.task()
def add(a, b):
    return a + b

result = add(1, 2)

Celery 与 Dramatiq 对比

Dramatiq：注重可靠性的现代替代方案

适用 Dramatiq 的场景：

• 可靠性优先于功能
• API 更简单
• 类型提示更完善
• 使用 RabbitMQ 或 Redis

适用 Celery 的场景：

• 成熟的生态系统
• 更多代理选项
• Canvas 工作流
• 更大的社区

# Dramatiq Example
import dramatiq

@dramatiq.actor
def add(x, y):
    return x + y

add.send(1, 2)

Celery 与云服务对比

AWS Lambda、Google Cloud Functions、Azure Functions

适用云函数的场景：

• 无服务器基础设施
• 事件驱动型工作流
• 按执行次数付费
• 自动扩缩容

适用 Celery 的场景：

• 自托管基础设施
• 复杂的任务工作流
• 成本可预测
• 完全控制执行过程

---

最佳实践

任务设计

1. 幂等性：任务应可安全地多次运行

   @app.task
   def process_order(order_id):
       order = Order.objects.get(id=order_id)
       if order.status == 'processed':
           return  # Already processed, skip
   
       order.process()
       order.status = 'processed'
       order.save()
   

2. 任务小而专注：每个任务只负责一项职责

   # Bad: Monolithic task
   @app.task
   def process_user(user_id):
       send_welcome_email(user_id)
       create_profile(user_id)
       setup_notifications(user_id)
   
   # Good: Separate tasks
   @app.task
   def send_welcome_email(user_id):
       ...
   
   @app.task
   def create_profile(user_id):
       ...
   
   workflow = group([
       send_welcome_email.s(user_id),
       create_profile.s(user_id),
       setup_notifications.s(user_id)
   ])
   

3. 避免在参数中传递数据库对象：改用 ID

   # Bad
   @app.task
   def process_user(user):  # User object
       ...
   
   # Good
   @app.task
   def process_user(user_id):
       user = User.objects.get(id=user_id)
       ...
   

4. 设置时间限制：防止任务失控

   @app.task(time_limit=300, soft_time_limit=240)
   def bounded_task():
       ...
   

错误处理

1. 对异常分类：针对不同错误采取不同的处理方式
2. 使用指数退避：避免给故障服务造成过大压力
3. 设置最大重试次数：不要无限重试
4. 记录失败信息：始终记录任务失败的原因

性能

1. 使用 ignore_result=True：适用于不需要结果的任务
2. 批量操作：每个任务处理多个项目
3. 优化序列化：使用 msgpack 提升速度
4. 连接池：重复使用数据库/代理连接
5. 任务分块：避免创建数百万个细小任务

监控

1. 启用事件：跟踪任务生命周期
2. 使用 Flower：基于 Web 的监控
3. 健康检查：监控工作进程是否可用
4. Sentry 集成：跟踪错误

安全

1. 验证输入：始终验证任务参数
2. 保护代理：使用身份验证和加密
3. 限制任务执行时间：防止资源耗尽
4. 速率限制：防止任务泛滥

---

故障排除

任务未执行

症状：任务已排队，但未处理

诊断：

# Check if workers are running
celery -A myapp inspect active

# Check worker stats
celery -A myapp inspect stats

# Check registered tasks
celery -A myapp inspect registered

解决方案：

• 启动工作进程：celery -A myapp worker
• 检查工作进程是否正在使用正确的队列
• 验证任务路由配置
• 检查代理连接

任务静默失败

症状：任务显示 SUCCESS，但实际未生效

诊断：

# Enable task tracking
app.conf.task_track_started = True

# Check task result and traceback
result = task.delay()
if result.failed():
    print(result.traceback)

解决方案：

• 检查日志：celery -A myapp worker --loglevel=debug
• 在测试中启用即时模式以查看异常
• 在测试中使用 task_eager_propagates = True

内存泄漏

症状：工作进程的内存使用量随时间增长

解决方案：

# Restart workers after N tasks
worker_max_tasks_per_child = 1000

# Restart on memory limit
worker_max_memory_per_child = 200000  # 200MB

任务执行缓慢

症状：任务耗时超出预期

诊断：

# Add timing
import time

@app.task(bind=True)
def timed_task(self):
    start = time.time()
    result = slow_operation()
    duration = time.time() - start
    logger.info(f"Task {self.request.id} took {duration}s")
    return result

解决方案：

• 提高工作进程并发数
• 优化任务代码
• 将任务拆分为多个块
• 添加更多工作进程

Broker 连接问题

症状：任务未到达工作进程

诊断：

# Test broker connection
python -c "from celery_app import app; print(app.connection().connect())"

解决方案：

• 检查 Broker 是否正在运行：redis-cli ping 或 rabbitmqctl status
• 验证配置中的 Broker URL
• 检查网络连接
• 启用连接重试：broker_connection_retry_on_startup = True

任务结果未持久化

症状：result.get() 返回 None

解决方案：

• 确认已配置结果后端
• 检查任务是否设置了 ignore_result=True
• 确认结果尚未过期（result_expires）
• 测试后端连接

Beat 未调度任务

症状：定期任务未运行

诊断：

# Check beat is running
ps aux | grep celery | grep beat

# Check beat schedule
celery -A myapp inspect scheduled

解决方案：

• 确保 Beat 进程正在运行
• 验证 beat_schedule 配置
• 检查 Beat 日志中的错误
• 对于动态计划，使用数据库调度器

工作进程崩溃

症状：工作进程意外退出

解决方案：

• 检查日志中的错误
• 设置 worker_max_tasks_per_child 以防止内存泄漏
• 添加任务时间限制
• 使用 systemd 自动重启
• 使用 Flower 进行监控

任务队列堆积

症状：任务在队列中不断堆积

解决方案：

• 添加更多工作进程
• 提高工作进程并发数
• 优化耗时较长的任务
• 添加任务路由以分散负载
• 检查是否有工作进程被阻塞

---

高级配置

自定义任务类

from celery import Task

class DatabaseTask(Task):
    """Task that manages database connections"""
    _db = None

    @property
    def db(self):
        if self._db is None:
            self._db = create_db_connection()
        return self._db

    def after_return(self, status, retval, task_id, args, kwargs, einfo):
        """Close connection after task"""
        if self._db is not None:
            self._db.close()

@app.task(base=DatabaseTask)
def db_task(query):
    return db_task.db.execute(query)

自定义序列化器

from kombu.serialization import register

def my_encoder(obj):
    # Custom encoding logic
    return json.dumps(obj)

def my_decoder(data):
    # Custom decoding logic
    return json.loads(data)

register('myjson', my_encoder, my_decoder,
         content_type='application/x-myjson',
         content_encoding='utf-8')

app.conf.task_serializer = 'myjson'

任务继承

class BaseTask(Task):
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        send_alert(f"Task {self.name} failed", str(exc))

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        logger.warning(f"Task {self.name} retrying")

@app.task(base=BaseTask)
def monitored_task():
    return perform_work()

---

Celery 技能文档结束

更多信息：

• 官方文档：https://docs.celeryq.dev/
• GitHub：https://github.com/celery/celery
• 社区：https://groups.google.com/forum/#!forum/celery-users

相关技能

使用 Celery 时，以下技能可增强你的工作流：

• django：Django + Celery 集成，用于后台任务
• fastapi-local-dev：用于异步 API 操作的 FastAPI + Celery 模式
• test-driven-development：测试异步任务和任务链
• systematic-debugging：调试分布式任务故障和竞态条件

[如果你的工具包中部署了这些技能，可查看完整文档]