每天40分玩转Django：Django Celery

Django Celery

一、知识要点概览表

模块	知识点	掌握程度要求
Celery基础	配置、任务定义、任务执行	深入理解
异步任务	任务状态、结果存储、错误处理	熟练应用
周期任务	定时任务、Crontab、任务调度	熟练应用
监控管理	Flower、任务监控、性能优化	理解应用

二、基础配置实现

1. 安装和配置

# requirements.txt
celery==5.3.1
django-celery-results==2.5.1
django-celery-beat==2.5.0
redis==4.6.0

# settings.py
INSTALLED_APPS = [
    ...
    'django_celery_results',
    'django_celery_beat',
]

# Celery配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'django-db'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'

# 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()

@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

三、异步任务实现

1. 基本任务定义

# tasks.py
from celery import shared_task
from django.core.mail import send_mail
from .models import User

@shared_task
def send_welcome_email(user_id):
    """发送欢迎邮件"""
    try:
        user = User.objects.get(id=user_id)
        send_mail(
            '欢迎加入我们的平台',
            f'你好 {user.username}，欢迎使用我们的服务！',
            'noreply@example.com',
            [user.email],
            fail_silently=False,
        )
        return True
    except Exception as e:
        return str(e)

@shared_task(bind=True, max_retries=3)
def process_payment(self, order_id):
    """处理支付任务"""
    from .models import Order
    try:
        order = Order.objects.get(id=order_id)
        result = process_payment_gateway(order)
        if result.success:
            order.status = 'paid'
            order.save()
            return 'Payment processed successfully'
        else:
            raise ValueError('Payment failed')
    except Exception as exc:
        self.retry(exc=exc, countdown=60*5)  # 5分钟后重试

2. 任务链和组

from celery import chain, group, chord
from .tasks import process_payment, send_notification, update_inventory

def process_order(order):
    # 任务链：按顺序执行任务
    task_chain = chain(
        process_payment.s(order.id),
        update_inventory.s(order.id),
        send_notification.s(order.user.id)
    )
    return task_chain()

def process_bulk_orders(orders):
    # 任务组：并行执行多个任务
    task_group = group(
        process_payment.s(order.id)
        for order in orders
    )
    return task_group()

def process_orders_with_summary(orders):
    # 和弦：并行执行任务后执行回调
    def on_complete(results):
        successful = sum(1 for r in results if r == 'success')
        failed = len(results) - successful
        return f"处理完成：{successful}成功，{failed}失败"
    
    task_chord = chord(
        (process_payment.s(order.id) for order in orders),
        on_complete.s()
    )
    return task_chord()

3. 自定义任务类

from celery import Task
from django.core.cache import cache

class BaseTaskWithRetry(Task):
    abstract = True
    max_retries = 3
    default_retry_delay = 60  # 60秒后重试

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        """任务失败时的处理"""
        print(f'Task {task_id} failed: {str(exc)}')
        super().on_failure(exc, task_id, args, kwargs, einfo)

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        """任务重试时的处理"""
        print(f'Task {task_id} retrying: {str(exc)}')
        super().on_retry(exc, task_id, args, kwargs, einfo)

@shared_task(base=BaseTaskWithRetry)
def process_data(data_id):
    try:
        # 处理数据的逻辑
        result = process_complex_data(data_id)
        return result
    except Exception as exc:
        raise self.retry(exc=exc)

四、周期任务实现

1. 基本周期任务

# tasks.py
from celery.schedules import crontab
from celery.task import periodic_task

@periodic_task(run_every=timedelta(hours=24))
def daily_cleanup():
    """每日清理任务"""
    cleanup_expired_tokens()
    cleanup_old_logs()
    return "Daily cleanup completed"

@periodic_task(
    run_every=crontab(hour=0, minute=0),
    name="generate_daily_report"
)
def generate_daily_report():
    """生成每日报告"""
    report = Report.objects.create(
        date=timezone.now(),
        type='daily'
    )
    report.generate()
    return f"Report generated: {report.id}"

2. 动态周期任务

# models.py
from django_celery_beat.models import PeriodicTask, IntervalSchedule
from django.db import models

class ScheduledReport(models.Model):
    name = models.CharField(max_length=100)
    interval = models.IntegerField(help_text='间隔（分钟）')
    enabled = models.BooleanField(default=True)
    
    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)
        self.update_periodic_task()
    
    def update_periodic_task(self):
        schedule, _ = IntervalSchedule.objects.get_or_create(
            every=self.interval,
            period=IntervalSchedule.MINUTES,
        )
        
        PeriodicTask.objects.update_or_create(
            name=f'generate_report_{self.id}',
            defaults={
                'task': 'myapp.tasks.generate_report',
                'interval': schedule,
                'args': [self.id],
                'enabled': self.enabled
            }
        )

3. 任务调度器

from django_celery_beat.models import CrontabSchedule, PeriodicTask
import json

def schedule_report_task(name, hour, minute, day_of_week):
    """创建定时报告任务"""
    schedule, _ = CrontabSchedule.objects.get_or_create(
        hour=hour,
        minute=minute,
        day_of_week=day_of_week,
    )
    
    task = PeriodicTask.objects.create(
        crontab=schedule,
        name=f'generate_report_{name}',
        task='myapp.tasks.generate_report',
        args=json.dumps([name]),
    )
    return task

# 使用示例
schedule_report_task('weekly_summary', hour=0, minute=0, day_of_week='1')  # 每周一

五、监控和管理

1. Flower配置

# requirements.txt
flower==2.0.1

# settings.py
CELERY_FLOWER_USER = 'admin'
CELERY_FLOWER_PASSWORD = 'password'

# 启动Flower
# celery -A myproject flower --port=5555

2. 任务监控中间件

# middleware.py
from celery.signals import task_prerun, task_postrun
import time
import logging

logger = logging.getLogger('celery.tasks')

@task_prerun.connect
def task_prerun_handler(task_id, task, args, kwargs, **kw):
    """任务执行前的处理"""
    task.start_time = time.time()

@task_postrun.connect
def task_postrun_handler(task_id, task, args, kwargs, retval, state, **kw):
    """任务执行后的处理"""
    if hasattr(task, 'start_time'):
        duration = time.time() - task.start_time
        logger.info(
            f'Task {task.name}[{task_id}] '
            f'completed in {duration:.2f}s with state {state}'
        )

六、Celery工作流程图

在这里插入图片描述

七、实际应用示例

1. 图片处理任务

# tasks.py
from PIL import Image
import os
from celery import shared_task

@shared_task
def process_uploaded_image(image_path, sizes=[(800, 600), (400, 300)]):
    """处理上传的图片"""
    try:
        img = Image.open(image_path)
        filename = os.path.basename(image_path)
        name, ext = os.path.splitext(filename)
        
        results = []
        for size in sizes:
            resized = img.copy()
            resized.thumbnail(size)
            new_filename = f"{name}_{size[0]}x{size[1]}{ext}"
            new_path = os.path.join(os.path.dirname(image_path), new_filename)
            resized.save(new_path)
            results.append(new_path)
        
        return results
    except Exception as e:
        return str(e)

2. 站点监控任务

# tasks.py
import requests
from celery import shared_task
from django.core.mail import mail_admins
from .models import SiteMonitor

@shared_task(bind=True, max_retries=3)
def monitor_website(self, url):
    """监控网站可用性"""
    try:
        response = requests.get(url, timeout=10)
        status = response.status_code == 200
        response_time = response.elapsed.total_seconds()
        
        SiteMonitor.objects.create(
            url=url,
            status=status,
            response_time=response_time
        )
        
        if not status:
            mail_admins(
                f'网站{url}不可用',
                f'状态码：{response.status_code}'
            )
            
        return {
            'url': url,
            'status': status,
            'response_time': response_time
        }
    except Exception as exc:
        self.retry(exc=exc, countdown=60)