Django 5.0 集成 Celery 5.4：异步任务实现指南

一、环境准备

1. 安装依赖包：

pip install django celery[redis] django-celery-beat django-celery-results

1. 项目结构：

myproject/
├── myproject/
│   ├── __init__.py
│   ├── celery.py   # 新增文件
│   ├── settings.py
│   └── urls.py
└── manage.py

二、核心配置

1. settings.py 配置：

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

# 定时任务扩展
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

# 注册应用
INSTALLED_APPS += [
    'django_celery_beat',
    'django_celery_results'
]

1. 创建 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()

1. 项目初始化文件 __init__.py：

from .celery import app as celery_app

__all__ = ('celery_app',)

三、实现异步任务

1. 创建任务文件 tasks.py：

from celery import shared_task
from django.core.mail import send_mail

@shared_task
def send_async_email(subject, message, recipient):
    send_mail(
        subject,
        message,
        'noreply@example.com',
        [recipient],
        fail_silently=False
    )
    return f"邮件已发送至 {recipient}"

1. 视图调用示例：

from .tasks import send_async_email

def trigger_email(request):
    send_async_email.delay(
        "订单确认",
        "您的订单已完成处理",
        "customer@example.com"
    )
    return HttpResponse("邮件任务已启动")

四、定时任务配置

1. 创建定时任务（在Django Admin操作）：

1. 访问 /admin/django_celery_beat/periodictask/add/
2. 选择任务：tasks.send_async_email
3. 设置调度规则（如每天9:00执行）

1. 代码创建定时任务：

from django_celery_beat.models import PeriodicTask, CrontabSchedule

schedule, _ = CrontabSchedule.objects.get_or_create(
    minute='0',
    hour='9',
    day_of_week='*',
    day_of_month='*',
    month_of_year='*'
)

PeriodicTask.objects.create(
    crontab=schedule,
    name='每日邮件报告',
    task='tasks.send_async_email',
    args=json.dumps([
        "每日报告",
        "这是自动发送的每日报告",
        "report@example.com"
    ])
)

五、运行服务

1. 启动Worker：

celery -A myproject worker --loglevel=info

1. 启动定时调度器：

celery -A myproject beat --loglevel=info

1. 监控任务结果：

# 查看任务状态
celery -A myproject status

# 查看定时任务
celery -A myproject beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

六、邮件发送配置

在 settings.py 中添加：

EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.example.com'
EMAIL_PORT = 587
EMAIL_USE_TLS = True
EMAIL_HOST_USER = 'your_email@example.com'
EMAIL_HOST_PASSWORD = 'your_password'

七、错误处理

1. 任务重试机制：

@shared_task(bind=True, max_retries=3)
def send_async_email(self, subject, message, recipient):
    try:
        # 邮件发送代码
    except SMTPException as exc:
        self.retry(exc=exc, countdown=60)  # 60秒后重试

八、最佳实践

1. 任务幂等性设计：确保重复执行不会产生副作用
2. 使用任务队列分离：为高优先级任务创建专用队列
3. 监控方案：
   • 使用Flower监控Celery：pip install flower
   • 启动：celery -A myproject flower
   • 访问：http://localhost:5555

注意：生产环境建议使用：

1. Supervisor管理进程
2. Redis持久化配置
3. TLS加密的消息传输

此方案完整实现了：

• 异步邮件发送
• 定时任务调度
• 任务状态追踪
• 错误自动恢复机制