学习在Django中集成Celery

pip install celery redis  # 使用 Redis 作为 Broker
pip install flower       # 可选，任务监控工具

your_project/
├── your_project/
│   ├── __init__.py
│   ├── celery.py       # 新增 Celery 初始化文件
│   ├── settings.py
│   └── urls.py
└── your_app/
    └── tasks.py        # 存放异步任务

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
app = Celery('your_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()  # 自动发现所有 Django App 中的 tasks.py

from .celery import app as celery_app
__all__ = ('celery_app',)

# Celery 配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'  # Broker 地址
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'  # 结果存储（可选）
CELERY_TIMEZONE = 'Asia/Shanghai'  # 时区设置

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

@shared_task
def add(x, y):
    return x + y

# 在视图或其他地方调用
from your_app.tasks import add

def register_user(request):
    # ... 用户注册逻辑
    add.delay(98, 98)  # 异步执行
    return HttpResponse("Registered!")

celery -A your_project worker --loglevel=info

celery -A your_project beat --loglevel=info

celery -A your_project flower  # 访问 http://localhost:5555

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-every-night': {
        'task': 'your_app.tasks.add',
        'schedule': crontab(hour=3, minute=30),  # 每天凌晨3:30执行
    },
}

@shared_task(bind=True, max_retries=3)
def process_data(self, data):
    try:
        # 处理数据...
    except Exception as e:
        self.retry(exc=e, countdown=60)  # 60秒后重试

from celery import chain
chain(task1.s(), task2.s(), task3.s()).apply_async()

result = add.delay(98, 98)
print(result.get(timeout=10))  # 获取任务结果（阻塞）

[program:celery_worker]
command=celery -A your_project worker --loglevel=info
directory=/path/to/your_project
autostart=true
autorestart=true

celery -A your_project worker --concurrency=4  # 根据 CPU 核心数调整