django-celery-expert

Django Celery 专家

说明

步骤 1：对请求进行分类

根据请求确定任务类别：

• Django 集成 — 事务安全、ORM 模式、测试、请求关联 → 阅读 references/django-integration.md
• 任务设计 — 新任务、调用模式、链／组／弦、幂等性 → 阅读 references/task-design-patterns.md
• 配置 — Broker 设置、结果后端、工作进程设置、队列路由 → 阅读 references/configuration-guide.md
• 错误处理 — 重试、退避、死信队列、超时 → 阅读 references/error-handling.md
• 周期性任务 — Celery Beat、crontab 计划、动态计划、时区处理 → 阅读 references/periodic-tasks.md
• 监控 — Flower、Prometheus、日志记录、调试卡住的任务 → 阅读 references/monitoring-observability.md
• 生产环境部署 — 扩缩容、进程监管、容器、运行状况检查 → 阅读 references/production-deployment.md

如果请求涉及多个类别，请先阅读所有相关参考文件，再继续。

步骤 2：阅读参考文件

阅读步骤 1 中确定的每个参考文件。未阅读相关参考文件之前，不要开始实施。

步骤 3：实施

应用参考文件中的模式。提交解决方案之前，请确认：

• 任务参数可序列化（传递 ID，而不是模型实例）
• 启用了重试的任务具有幂等性
• 错误已附带上下文记录
• 长时间运行的任务已配置超时

示例

基本后台任务

请求：“用户注册后在后台发送欢迎邮件”

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

@shared_task(bind=True, max_retries=3)
def send_welcome_email(self, user_id):
    from users.models import User

    try:
        user = User.objects.get(id=user_id)
        send_mail(
            subject="Welcome!",
            message=f"Hi {user.name}, welcome to our platform!",
            from_email="[email protected]",
            recipient_list=[user.email],
        )
    except User.DoesNotExist:
        pass
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))


# views.py — queue only after the transaction commits
from django.db import transaction

def register(request):
    user = User.objects.create(...)
    transaction.on_commit(lambda: send_welcome_email.delay(user.id))
    return redirect("dashboard")

带进度跟踪的任务

请求：“处理大型 CSV 导入并提供进度更新”

@shared_task(bind=True)
def import_csv(self, file_path, total_rows):
    from myapp.models import Record

    with open(file_path) as f:
        reader = csv.DictReader(f)
        for i, row in enumerate(reader):
            Record.objects.create(**row)
            if i % 100 == 0:
                self.update_state(
                    state="PROGRESS",
                    meta={"current": i, "total": total_rows},
                )

    return {"status": "complete", "processed": total_rows}


# Poll progress
result = import_csv.AsyncResult(task_id)
if result.state == "PROGRESS":
    progress = result.info.get("current", 0) / result.info.get("total", 1)

使用链的工作流

请求：“处理订单：验证库存、收取款项，然后发送确认信息”

from celery import chain

@shared_task
def validate_inventory(order_id):
    order = Order.objects.get(id=order_id)
    if not order.items_in_stock():
        raise ValueError("Items out of stock")
    return order_id

@shared_task
def charge_payment(order_id):
    order = Order.objects.get(id=order_id)
    order.charge()
    return order_id

@shared_task
def send_confirmation(order_id):
    Order.objects.get(id=order_id).send_confirmation_email()

def process_order(order_id):
    chain(
        validate_inventory.s(order_id),
        charge_payment.s(),
        send_confirmation.s(),
    ).delay()