企业官网建设流程全解析

django-tasks 实践教程：给 Django 加后台任务队列、延迟任务与结果查询

【免费下载链接】django-tasksA backport of Django's built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks

当 Django 项目需要发邮件、生成报表、清理过期数据时，同步执行会拖慢响应。django-tasks 让你用一个装饰器把普通函数变成任务，通过一次enqueue调用派发，并支持延迟执行、状态查询与错误取回。下面从最小配置开始，把任务队列接入已有项目。

什么时候需要引入 django-tasks

几个典型信号：

• 请求里存在不影响页面主流程的重活：发通知邮件、导出文件、重建缓存；
• 希望把某个动作推迟到稍后执行，比如"一小时后发送提醒"；
• 发起方需要知道后台操作最终成功与否，并能取回返回值或异常。

django-tasks 是 Django 官方内置任务框架的向后移植，API 与官方方向一致。核心概念只有一个：任务后端决定任务是在当前进程执行，还是交给外部系统执行。框架自带两个后端，不需要额外依赖就能跑起来。

让任务队列跑起来：安装与最小配置

先安装包，一条命令即可：

python -m pip install django-tasks

然后在settings.py中启用应用并声明后端。TASKS的写法类似DATABASES，支持多个别名，后端选择见 后端源码：

INSTALLED_APPS = [ # ... "django_tasks", ] TASKS = {"default": {"BACKEND": "django_tasks.backends.immediate.ImmediateBackend"}}

ImmediateBackend在当前线程同步执行任务：enqueue返回时任务已经跑完。它适合本地开发和"暂时不需要真异步、但想统一接口"的阶段。不配置TASKS时框架会回退到同样的默认值，但显式写出来更清楚。

派发第一个后台任务

任务是加了@task装饰器的普通函数。函数必须定义在模块顶层，不能是闭包或内部函数——后端序列化任务时靠模块路径找回它。

以"发送欢迎邮件"为例，这是最常见的真实场景：

from django_tasks import task @task() def send_welcome_email(to: str, subject: str) -> None: # 发送邮件的逻辑

调用enqueue派发，位置参数和关键字参数会原样传给函数，返回值是一个TaskResult：

result = send_welcome_email.enqueue("user@example.com", "欢迎加入") # 派发时按次调整优先级和队列，不影响原始定义 send_welcome_email.using(priority=10, queue_name="reports").enqueue(to, subject)

using返回一个新的 Task 实例，相当于"复制一份并改掉默认值"。在异步视图或协程里，用await task.aenqueue(...)代替enqueue。

控制执行时机与结果

配置延迟任务

延迟执行通过run_after指定最早执行时间。注意两点：项目开启USE_TZ时必须传带时区的datetime；而且延迟能力取决于后端——默认的ImmediateBackend不支持run_after，直接派发会抛InvalidTask。

一小时后再执行，写法如下：

from datetime import timedelta from django.utils import timezone send_welcome_email.using(run_after=timezone.now() + timedelta(hours=1)).enqueue(to, subject)

派发前可用default_task_backend.supports_defer判断当前后端是否支持，配合 Django 的 checks 框架能在启动时暴露配置问题。同理还有supports_priority（优先级）、supports_async_task（async 任务函数）、supports_get_result（跨线程/进程取回结果）四个布尔标志。

查询任务状态与结果

TaskResult的status有四种取值：READY（已入队未执行）、RUNNING、SUCCESSFUL、FAILED。返回值通过return_value属性获取，前提是状态为成功，否则会抛ValueError；失败信息在errors列表里（当前实现只会有一个元素）：

if result.status is TaskResultStatus.SUCCESSFUL: value = result.return_value elif result.status is TaskResultStatus.FAILED: print(result.errors[0].exception_class) # 异常类型 print(result.errors[0].traceback) # 完整堆栈字符串

如果任务在后台被更新过，调用result.refresh()从存储重新加载。跨请求取回结果时，把result.id当作不透明字符串保存（最长 64 字符），之后用send_welcome_email.get_result(result_id)取回（任务类型不匹配会抛TaskResultMismatch），或用default_task_backend.get_result(result_id)取回任意任务——后一种方式只对声明了supports_get_result的后端有意义。

选择执行后端与队列划分

两个内置后端的分工

• ImmediateBackend：enqueue即执行，结果只存在于当前进程内存中，重启即失。适合开发和演示。
• DummyBackend：只记录、不执行，入队后状态停在READY，结果存进后端的results列表（当前线程内可见，可用clear()清空）。它是单元测试的主力。

生产环境的分布式执行需要自行实现BaseTaskBackend，或选用社区后端（django-tasks-db、django-tasks-rq自 0.12 起不再随主包分发）。

用 QUEUES 约束队列

默认所有任务进入default队列。多队列时建议在配置中显式列出允许的名称，防止任务派发到没人消费的队列：

TASKS = {"default": {"BACKEND": "...", "QUEUES": ["default", "reports"]}}

派发到未知队列会抛InvalidTask；把QUEUES设为空列表则关闭校验。指定队列有两个时机：定义时@task(queue_name="reports")固定，或派发时.using(queue_name="reports")临时切换。

任务测试、监控与故障定位

用 DummyBackend 验证任务

单测中换成DummyBackend，既不碰真实邮件等外部资源，又能断言"任务确实被派发、参数正确"。参考仓库里的 测试任务示例：

@override_settings(TASKS={"default": {"BACKEND": "django_tasks.backends.dummy.DummyBackend"}}) class WelcomeEmailTests(SimpleTestCase): def test_enqueued(self): result = send_welcome_email.enqueue("u@example.com", "hi") self.assertEqual(result.status, TaskResultStatus.READY)

result.args、result.kwargs记录实际参数，result.attempts记录执行次数（当前实现只有 0 或 1）。

失败处理与长任务监控

框架提供三个信号：task_enqueued、task_started、task_finished，sender是后端类，task_result以关键字参数传入。框架已内置接收器，会把任务生命周期写入django_tasks日志器，失败的日志级别会升级为exception。想接入告警时挂自己的接收器：

@receiver(task_finished) def handle_finished(sender, task_result, **kwargs): if task_result.status is TaskResultStatus.FAILED: error_logger.error(task_result.errors[0].traceback)

监控长任务时，内置后端没有"查询所有运行中任务"的接口。务实的做法：派发时把result.id记入自己的数据表，定期用get_result轮询（需后端支持），检查started_at、last_attempted_at和is_finished，超时未完成的再人工介入。

边界在哪：django-tasks 不做什么

• 没有取消 API。任务一旦派发无法外部撤回，变通方式是在任务函数里检查业务标志（比如takes_context=True拿到的上下文）自行决定中止。
• 结果留存取决于后端。内置两个后端的结果都在内存里，重启即失；需要长期留存就在任务结束时自行落库。
• 参数与返回值会做 JSON 归一化，不可序列化的对象会被转成字符串存储，别指望取回原对象。
• 内置后端不做并行。ImmediateBackend的"后台任务"其实就是同步函数，长任务会阻塞请求本身；真正的分布式调度、失败自动重试，需要换用带 worker 的后端。

把"重活"从请求路径剥离，是 django-tasks 解决的第一件事。接口统一后，换后端只改TASKS配置，任务代码不动。

【免费下载链接】django-tasksA backport of Django's built in Tasks framework项目地址: https://gitcode.com/gh_mirrors/dj/django-tasks