企业官网建设流程全解析

django-tasks 配置清单：5 分钟让 Django 异步任务从零到生产

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

django-tasks 是 Django 官方任务框架的向后移植：在 Django 5.2 上，用官方的 @task 装饰器、enqueue 入队与结果追踪处理 Django 异步任务、定时任务与任务队列，无需引入 Celery。

为什么是任务框架，而不是 Celery

电商后台里，注册流程同步发验证邮件，商品页同步压缩图片，订单超时清理也搭在用户请求里执行——接口响应从几十毫秒涨到几秒。

这类慢操作有两个出口：要么引入 Celery 加一整套 broker 运维，要么用 Django 官方自带的任务框架。django-tasks 就是官方框架的向后移植，让没到该版本的 Django 项目直接用上这套原生任务系统。

写法和官方文档一致，以后升级版本只需把导入路径换成django.tasks，业务代码原样保留。

5 分钟配置清单：跑通第一个任务

装包、注册、配后端三步走完，就可以发出第一个任务。

python -m pip install django-tasks

下面这段在settings.py里完成注册与后端选择：

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

配置不写也不报错——默认值就是 ImmediateBackend。各配置项的作用如下：

然后在任务模块里定义并执行第一个任务：

from django_tasks import task @task() def compress_product_image(image_id: int) -> str: # 压缩商品图，返回产物路径 ... result = compress_product_image.enqueue(1024) print(result.status) # TaskResultStatus.SUCCESSFUL

enqueue()的返回值TaskResult是后续一切状态追踪的入口，下一节展开。

@task 装饰器的四种写法

任务本质是"一个模块级函数 + 一组执行参数"，装饰器参数决定它去哪儿跑、以什么顺序跑。

不带参数最简单，等价于默认值：

@task def send_verification_email(address: str) -> None: ...

带参数形式覆盖优先级、队列与后端：

@task(priority=10, queue_name="media", backend="default") def compress_product_image(image_id: int) -> str: ...

不想改定义时，用using()生成一份覆盖默认值的副本，原任务不受影响：

urgent = compress_product_image.using(priority=80, queue_name="media") urgent.enqueue(2048)

带上下文时，函数第一个参数必须命名为context：

from django_tasks import task, TaskContext @task(takes_context=True) def generate_weekly_report(context: TaskContext, year: int, week: int) -> str: print(f"attempt={context.attempt}") ...

两个约束：任务函数必须定义在模块级，嵌套函数会在装饰器阶段直接抛InvalidTask；context提供task_result（当前TaskResult）与attempt（尝试次数），需要任务内部自报状态时用。

执行与结果怎么查

enqueue()把任务交给后端：Immediate 后端当场执行完才返回；真正的队列后端则先落队列、稍后执行，此时拿到的是READY状态的结果。

result = send_verification_email.enqueue("user@example.com") # 指定时刻执行；后端不支持延迟时会被拒绝 deferred = generate_weekly_report.using(run_after=planned_time).enqueue(2026, 34)

状态、返回值与错误都挂在TaskResult上：

from django_tasks import TaskResultStatus if result.status == TaskResultStatus.SUCCESSFUL: print(result.return_value) elif result.status == TaskResultStatus.FAILED: print(result.errors[0].exception_class) print(result.errors[0].traceback)

四个状态是READY（入队待执行）、RUNNING（执行中）、SUCCESSFUL、FAILED。注意return_value只在SUCCESSFUL时可用，失败或未完成时访问会抛ValueError。

跨请求或跨任务要拿结果，先存result.id，之后按 id 取回：

result_id = result.id # 长度不超过 64 的字符串，存库即可 # 任务侧取回，id 与任务不匹配会抛 TaskResultMismatch later = generate_weekly_report.get_result(result_id) # 后端侧取回任意任务的结果 from django_tasks import default_task_backend any_result = default_task_backend.get_result(result_id)

结果对象是快照：后台状态变了就调result.refresh()刷新，is_finished可快速判断是否已终态。

监控卡在RUNNING的长任务时，框架没有现成的批量查询接口，稳妥做法是自己落一份 id 台账，周期任务里逐个get_result再按started_at判断是否超时：

for rid in pending_ids: r = default_task_backend.get_result(rid) if r.status == TaskResultStatus.RUNNING and r.started_at < now - one_hour: flag_stuck(r)

后端与队列怎么搭配

两个内置后端解决的是两个不同问题：一个让功能先跑起来，一个让测试不依赖执行。

用supports_*标志在运行时确认能力，避免"部署后才发现功能不可用"：

from django_tasks import default_task_backend if default_task_backend.supports_defer: scheduled = generate_weekly_report.using(run_after=planned_time)

四个标志的含义：supports_defer（能否run_after延迟执行）、supports_async_task（能否入队协程）、supports_get_result（能否从任意线程/进程按 id 取结果）、supports_priority（能否按优先级排序执行）。

⚠️ 两个内置后端都覆盖不了"跨请求真异步"：Immediate 在请求线程里同步跑，慢任务照样拖长响应。生产环境需要搭配真正基于队列的后端（db 或 RQ 系扩展，导入路径见django_tasks/backends/base.py的BaseTaskBackend），选型时先查这四个标志。

队列名是逻辑分组：QUEUES写白名单后，未列出的队列名在装饰或入队时抛InvalidTask，防止打错字导致任务"消失"；设为[]则关闭校验。典型分法：

TASKS = { "default": { "BACKEND": "django_tasks.backends.immediate.ImmediateBackend", "QUEUES": ["default", "media", "reports"], }, }

IO 密集任务（压缩图片、发邮件）放media，计算任务（周报）放reports，慢的挤不占快的。

用信号监听任务生命周期

三个信号对应任务的完整生命周期：task_enqueued（入队后）、task_started（执行前一刻）、task_finished（执行结束，无论成败）。handler 以关键字参数收到task_result：

from django.dispatch import receiver from django_tasks import TaskResultStatus from django_tasks.signals import task_finished @receiver(task_finished) def on_task_finished(sender, task_result, **kwargs): if task_result.status == TaskResultStatus.FAILED: alert(task_result.task.module_path, task_result.errors[0].traceback)

sender是后端类，可以据此区分不同后端；框架自身已内置了打日志的 handler，失败任务会走logger.exception带出 traceback。埋点、通知、统计都挂在这里，不用改业务代码。

上生产前的避坑清单

最后对照这十条过一遍，能少掉进大多数坑：

• 任务函数定义在模块级，嵌套函数抛InvalidTask
• run_after仅对supports_defer的后端有效，否则任务立即执行
• 结果保留时长取决于后端：Immediate 的结果随进程结束消失，长保留需换支持持久化的后端
• 没有取消 API，已入队任务撤不回，靠任务内部检查取消标记实现业务级取消
• 查失败任务看errors[0].exception_class与errors[0].traceback
• return_value在FAILED/未完成时抛ValueError，访问前先判status
• QUEUES是白名单而非注册表，写空列表[]才关闭校验
• priority只能是 -100 到 100 的整数，后端不支持时设非零值直接抛错
• 长任务自己落 id 台账 + 周期get_result做超时巡检
• attempts当前取值只有 0 或 1，不要拿它当重试计数依赖

官方任务框架的 API 边界很克制：定义、入队、追踪、信号，四件事闭环。把 django-tasks 当底座，慢操作移出请求链路，剩下的交给队列。

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