django-q2

Django Q2 后台任务

在修改任务入队、计划、工作进程部署、Q_CLUSTER 或 Django Q2 工作进程导入的代码之前，请先阅读本节。

配置检查

• 依赖项是 django-q2；Python 导入路径是 django_q。
• django_q 位于 INSTALLED_APPS 中；其迁移会提供任务结果、计划和代理模型。
• 修改任务、工作进程或代理行为之前，请先在项目的设置模块中找到 Q_CLUSTER。
• 确认已配置的代理。Redis 很常见，通常通过 REDIS_URL 之类的环境变量配置；对于低吞吐量或不使用 Redis 的部署，ORM 代理很实用。
• 确认项目的工作进程命令。基础命令是 python manage.py qcluster，但项目可能会使用 uv、Poetry、Docker Compose、进程管理器或特定平台的工作进程声明进行封装。
• 如果移除 Redis 作为代理，请另外检查所有依赖 Redis 的缓存、运行状况检查、Docker、部署和文档引用。

心智模型

• Web 代码会调用 async_task(...) 或创建 Schedule 行。
• 代理会存储排队的任务包，直到 qcluster 进程预留任务。
• 工作进程会执行可导入的 Python 函数，并保存结果或失败信息。
• 调度器在集群中运行。计划是数据库行；除非 qcluster 正在运行，否则它们不会执行任何操作。

添加任务

1. 将任务函数放在拥有相应行为的应用中，通常是 tasks.py 模块，或项目已使用的其他可导入模块。
2. 确保函数在模块导入时可导入。不要依赖请求对象、局部闭包或进程本地状态。
3. 传递主键等持久标识符，而不是模型实例、打开的文件、连接或大型负载。
4. 确保任务具有幂等性。Redis 不保证恰好执行一次；基于回执的代理可能会重新运行任务。
5. 如果任务依赖刚保存的数据库行，请通过 transaction.on_commit(...) 将其加入队列。

def send_welcome_email(user_id: int) -> None:
    from django.contrib.auth import get_user_model

    user = get_user_model().objects.get(pk=user_id)
    ...

from django.db import transaction
from django_q.tasks import async_task

transaction.on_commit(
    lambda: async_task("myapp.tasks.send_welcome_email", user.pk)
)

当 Django Q2 选项可能与任务关键字参数冲突时，请使用 q_options：

async_task(
    "myapp.tasks.rebuild_report",
    report_id,
    q_options={"timeout": 300, "group": "reports"},
)

计划任务

优先使用通过迁移、管理操作或设置命令创建的命名幂等计划。避免在导入时或应用启动时无条件创建计划。

from django_q.models import Schedule

Schedule.objects.get_or_create(
    name="clear-expired-sessions",
    defaults={
        "func": "django.core.management.call_command",
        "args": "'clearsessions'",
        "schedule_type": Schedule.HOURLY,
    },
)

使用 Schedule.objects.get_or_create(name=..., defaults={...}) 初始化计划，避免重复运行设置操作时产生重复任务。Cron 计划需要可选依赖 croniter；如果项目未包含该依赖，请勿使用 Schedule.CRON。

默认情况下，错过的计划会补执行。如果任务应在停机后仅运行一次，而非重放每个错过的间隔，请设置 Q_CLUSTER["catch_up"] = False。

代理选择

Redis 代理

如果项目已在工作进程或部署中依赖 Redis，请使用 Redis：

Q_CLUSTER = {
    "name": "...",
    "timeout": 3600,
    "workers": 4,
    "redis": REDIS_URL,
}

Redis 速度快，通常适合已为缓存、Docker 或部署工作进程运行 Redis 的项目。默认 Redis 代理不支持投递回执。如果工作进程主机在执行任务时发生灾难性故障，正在处理的任务包可能会丢失；如果任务代码引发异常，Django Q2 会记录失败。请采用幂等任务设计，在需要时于任务代码中显式重试，并监控失败情况。

ORM Broker

仅在低吞吐量部署、追求本地简洁性或 Redis 不可用的环境中使用 Django 数据库 Broker：

Q_CLUSTER = {
    "name": "...",
    "timeout": 3600,
    "retry": 4800,
    "workers": 4,
    "max_attempts": 2,
    "orm": "default",
}

切换到 ORM 时：

• 移除 "redis" Broker 键；每个集群配置一个 Broker，除非你有意使用自定义集群。
• 为 django_q 运行迁移。如果 Broker 使用非默认数据库别名，请使用 --database <alias> 运行迁移。
• 如果需要降低数据库轮询压力，且可以接受更高的队列任务获取延迟，请将 "poll" 调高至默认值 0.2 秒以上，例如 "poll": 2.0。
• ORM Broker 会启用 Queued Tasks 管理表。
• 单独检查依赖 Redis 的缓存、健康检查、Docker 和部署设置。计划任务始终存储为数据库行；Broker 设置控制的是排队任务包，而非计划任务表。

测试

• 直接调用函数，测试任务业务逻辑。
• 使用同步模式测试任务入队：
  • 每次调用：async_task("myapp.tasks.fn", arg, sync=True)
  • 每个测试：覆盖 Q_CLUSTER["sync"] = True
• 测试 Worker/Broker 集成时，在单独进程中运行项目的 qcluster 命令，并等待 result(task_id, 200) 或类似的有界等待；不要依赖任意时长的休眠。
• 当真实 Worker 进程必须读取已提交的数据库行时，使用 pytest.mark.django_db(transaction=True)。

调试检查清单

• 是否有一个 qcluster 进程正在运行，并且使用与 Web 进程相同的设置模块、SECRET_KEY、Broker URL 和集群名称？
• Worker 能否导入点分隔的任务路径？
• 是否运行了数据库迁移，包括 django_q 迁移？
• Web 和 Worker 容器是否都能访问 Redis，或者 ORM Broker 是否正在轮询预期的数据库？
• 设置过程是否在没有稳定名称的情况下创建了另一个 Schedule 行，从而导致计划任务重复？
• 停机是否触发了计划任务补执行？
• 能否在 Django 管理后台、日志或配置的错误报告器中看到任务失败？

参考资料

• 官方文档：https://django-q2.readthedocs.io/en/master/
• 上游仓库：https://github.com/django-q2/django-q2