首次探秘 Django 的新后台任务框架

Django 6.0 在 django.tasks 中引入了内置的后台任务框架。但请勿急于淘汰 Celery、Huey 或其他首选解决方案。

发布说明对此有明确说明：

Django 负责任务创建和队列管理，但 不提供执行任务的工件机制 。任务执行必须由外部基础设施(如独立进程或服务)管理。

新版 django.tasks 模块的核心目标是 为任务队列实现提供通用 API 。Jake Howard 是这项增强功能的核心推动者。欢迎查阅Django论坛上的项目介绍。

他的参考实现(同时兼容早期Django版本的回溯移植)已发布于GitHub上的django-tasks。

但我们暂且忽略该实现，转而探索Django 6.0内置的精简版本。通过创建专属后端和工作进程，我们来实践：

我们的项目：通知系统

我们将创建一个应用程序，利用ntfy.sh向手机及其他设备发送通知。(我是它的粉丝！)

若想直接参与代码开发，请查看GitHub上的最终项目版本。

使用nfty向手机发送通知只需：

1. 注册账户
2. 创建主题。
3. 在手机上安装应用并登录。
4. 向 https://ntfy.sh/<yourtopic> 发送HTTP请求

免费版仅提供公开主题和消息。这意味着任何订阅该主题的人都能看到您发送的内容。为满足我们的需求，可直接创建随机命名的主题(如UUID)。

项目配置要求将步骤4的URL作为环境变量提供。例如：

NTFY_URL=https://ntfy.sh/062519693d9c4913826f0a39aeea8a4c

以下是承担核心功能的函数：

import httpx
from django.conf import settings

def send_notification(message: str, title: str | None):
    # Pass the title if specified.
    headers = {"title": title} if title else {}
    httpx.post(
        settings.NTFY_URL,
        content=message,
        headers=headers,
    )

真的。至此已具备发送接收通知的基本功能。

快速入门指南

建议查阅Django任务框架文档获取详细说明，但为节省时间我们提供简要指南。

定义任务

这是新框架的核心目标：通过 Django 标准 API 定义任务，而非使用任务队列专属装饰器或其他方法。

具体实现如下：

# ...
from django.tasks import task

@task
def send_notification(message: str, title: str | None):
    # ...as before

该函数现已成为任务。准确来说是 django.tasks.Task 对象。

您无法再直接调用send_notification。任务必须通过enqueue方法运行。这可能与您的预期行为不符，但这是最佳方案。该设计彻底杜绝了在进程中意外调用任务(而非后台执行)的可能性。

task装饰器允许您指定任务优先级、队列名称和后端名称。可通过using方法覆盖这些设置，该方法将返回新的django.tasks.Task实例。

若需更精细地控制任务行为，可在装饰器中将takes_context设为True，并将context作为首参数传递。当前该上下文可访问任务结果，从而获取尝试次数等有用信息。

无法定义重试机制、延迟策略等完整任务队列实现中的高级功能——但这 并非 该组件的设计初衷。如有需要，您可通过检查任务上下文轻松添加自定义重试逻辑。

任务入队

向队列添加任务非常简单：

task_result = send_notification.enqueue(
    message="Season's greeting!", 
    title="Santa has something to tell you"
)

执行任务

此处功能尚有不足。至少当前版本如此。Django 6.0 将提供 ImmediateBackend 和 DummyBackend 两种后端：前者立即执行任务，后者则完全不执行任务。

因此我们的项目包含一个基于数据库和工作进程的(演示)后端！

获取结果

若无需即时等待结果，可通过任务ID后续获取。只需调用任务的get_result(result_id)方法即可。

本项目包含一个定期通过htmx轮询未处理结果的视图。

表单下方的列表展示每次任务执行的结果。表单提交后，新结果将添加至列表顶部。只要结果状态未变为FAILED或SUCCESSFUL，Htmx就会持续轮询更新。

def task_result(request, result_id, status):
    result = send_notification.get_result(result_id)
    if result.status == status:
        # No need to swap the result.
        return HttpResponse(status=204)
    return TemplateResponse(request, "index.html#result", {"result": result})

想知道index.html#results在做什么？Django 6.0还引入了模板片段。在此示例中，我们的视图实际上发送的响应仅包含名为result的模板片段。

幕后机制

当你用 task 装饰器修饰可调用对象时，系统会调用配置后端的 task_class 类来封装该对象。默认使用 django.task.Task 类。

该类的 enqueue 方法会进一步调用配置后端的 enqueue 方法。

调用其get_result方法的逻辑类似：调用配置后端的get_result方法并传递结果。

由于无需工作进程，这基本就是任务后端所需实现的全部功能。很酷吧？我们来添加一个后端吧？

任务数据库后端

我们的目标：

• 基于数据库实现基础任务后端
• 支持“自动重试”机制

我们的enqueue和get_result方法将返回默认django.tasks.TaskResult实例。这决定了我们需要存储的最小数据量，我们将通过名为Task的模型实现存储。

模型设计

基于 django.tasks 中 TaskResult 和 Task 的属性(即“数据类”)，我们先创建 Task 模型的初稿：

class Task(models.Model):
    priority = models.IntegerField(default=0)
    callable_path = models.CharField(max_length=255)
    backend = models.CharField(max_length=200)
    queue_name = models.CharField(max_length=100)
    run_after = models.DateTimeField(null=True, blank=True)
    takes_context = models.BooleanField(default=False)
    # Stores args and kwargs
    arguments = models.JSONField(null=True, blank=True)
    status = models.CharField(
        choices=TaskResultStatus.choices, max_length=10, default=TaskResultStatus.READY
    )
    enqueued_at = models.DateTimeField()
    started_at = models.DateTimeField(blank=True, null=True)
    finished_at = models.DateTimeField(blank=True, null=True)
    last_attempted_at = models.DateTimeField(blank=True, null=True)
    return_value = models.JSONField(null=True, blank=True)

缺失了什么？首先，TaskResult还包含遇到的错误列表以及处理任务的工人的ID。这些信息我们 或许 可以忽略。

但TaskResult.attempts属性基于工人ID的数量。若在任务内部使用任务上下文，必然需要依赖此类信息。

我们可通过为每个字段添加JSONField将这些细节纳入Task模型——这正是参考实现中的当前做法。

但让我们更明确地定义模型：记录每次任务执行尝试及其潜在错误，并通过外键与任务关联：

class Error(models.Model):
    exception_class_path = models.TextField()
    traceback = models.TextField()

class AttemptResultStatus(TextChoices):
    # Subset of TaskResultStatus.
    FAILED = TaskResultStatus.FAILED
    SUCCESSFUL = TaskResultStatus.SUCCESSFUL

class Attempt(models.Model):
    task = models.ForeignKey(Task, related_name="attempts", on_delete=models.CASCADE)
    error = models.OneToOneField(
        Error, related_name="attempt", on_delete=models.CASCADE, null=True, blank=True
    )
    worker_id = models.CharField(max_length=MAX_LENGTH_WORKER_ID)
    started_at = models.DateTimeField()
    stopped_at = models.DateTimeField(blank=True, null=True)
    status = models.CharField(
        choices=AttemptResultStatus.choices, max_length=10, blank=True
    )

此架构确保我们拥有执行任务所需的所有信息，同时在请求TaskResult时能提供完整细节。

一切看似完美，但我们还需考虑工作进程的需求。它必须能够：

1. 快速检查待处理任务
2. 领取其中一项任务
3. 处理该任务并标记为失败、成功或准备就绪 (以便稍后重试)

虽然现有架构能实现上述功能，但我希望进一步优化设计。

class Task(models.Model):
    # ...
    # This field is used to keep track of when to run a task (again).
    # run_after remains unchanged after enqueueing.
    available_after = models.DateTimeField()
    # Denormalized count of attempts.
    attempt_count = models.IntegerField(default=0)
    # Set when a worker starts processing this task.
    worker_id = models.CharField(max_length=MAX_LENGTH_WORKER_ID, blank=True)
    # ...

available_after字段将存储任务最早可执行时间。若任务已指定run_after(可通过任务的… using()方法实现)，则available_after将设为此值。否则默认采用当前日期时间(均以UTC为准)。

当任务需重试时，available_after将设为任务可执行的下一个时间点。换言之：我们可实施 延迟策略 。

attempt_count字段能简化可用任务的查询流程。任何attempt_count超过最大允许值的任务均可忽略。虽然其状态本应设置为FAILED(默认应被排除)，但我们仍可通过修改配置调整最大重试次数。

当工作者申领任务时，worker_id字段将被填充。这能防止其他工作者拾取该任务(前提是工作者ID具有唯一性)。

任务入队与结果获取

任务入队操作极其简单：从Task数据类实例创建Task模型实例，保存即可！当然，这需要在最终结果转换为TaskResult后完成。

我们使用模型数据库ID的字符串版本作为结果ID。

获取结果同样只需加载任务及其尝试记录，并转换为TaskResult即可。

以下是当前任务后端的简化版本：

class DatabaseBackend(BaseTaskBackend):
    supports_defer = True
    supports_async_task = False
    supports_get_result = True
    supports_priority = True

    def enqueue(self, task: Task, args, kwargs):
        self.validate_task(task)
        model = self.queue_store.enqueue(task, args, kwargs)
        task_result = TaskResult(
            task=task,
            id=str(model.pk),
            # ...
            # More properties being set
            # ...
        )
        return task_result

    def get_result(self, result_id):
        return self.model_to_result(
            self.queue_store.get(result_id)
        )

    def model_to_result(self, model: models.Task) -> TaskResult:
        ...

大量功能依赖于queue_store属性实现。在深入探讨前，我们将先说明该后端的配置选项。

配置

我们需要为以下项设置默认值：

• 最大尝试次数(重试次数)
• 退避因子；即使用 math.pow(factor, attempts) 实现退避机制

这些参数可针对每个队列单独定制。因此我们在 OPTIONS 中最终得到如下配置：

TASKS = {
    "default": {
        "BACKEND": "messagecenter.dbtasks.backend.DatabaseBackend",
        "OPTIONS": {
            "queues": {
                "low_priority": {
                    "max_attempts": 5,
                }
            },
            "max_attempts": 10,
            "backoff_factor": 3,
            "purge": {"finished": "10 days", "unfinished": "20 days"},
        },
    }
}

添加到 low_priority 队列的任务最多尝试五次，退避因子为 3。其他任务最多尝试十次，采用相同的退避因子。

队列存储

QueueStore类是后端的配套组件，专注于任务检索与入队、执行任务检查及任务申领。

然而将其纳入的主要目的是简化工人进程。正如我们将看到的，工人进程会获得队列存储的独立副本， 仅限于其需要处理的队列 。

工人进程

至少在本项目中，工人进程的职责是向运行器提供待处理任务的信息，并驱动后端对这些任务进行处理。其实现逻辑如下：

class Worker:
    def __init__(
        self,
        id_: str | None,
        backend_name: str,
        only: set[str] | None,
        excluding: set[str] | None,
    ):
        # Grab the backend and its queue_store.
        self.backend = task_backends[backend_name]
        queue_store: QueueStore = self.backend.queue_store
        # Limit the queue_store to the select queues.
        if only or excluding:
            queue_store = queue_store.subset(only=only, excluding=excluding)
        self.queue_store = queue_store
        # Use or create and id. "Must" be unique.
        self.id = (
            id_ if id_ else create_id(backend_name, queues=queue_store.queue_names)
        )

    def has_more(self) -> bool:
        return self.queue_store.has_more()

    def process(self):
        with transaction.atomic():
            tm = self.queue_store.claim_first_available(worker_id=self.id)
        if tm is not None:
            self.backend.process_task(tm)

要使工作者运行器正常运作，只需完成以下步骤：

1. 创建工作者实例。
2. 通过has_more查询是否有待执行任务。
3. 若有：指令其process处理首个可用任务。若无：转至步骤4。
4. 等待，然后返回步骤 2。

这正是我们的 dbtasks_worker 命令 的工作原理。

任务申领

队列存储提供peek方法，该方法返回队列中紧急程度最高的任务ID(综合考虑available_after、priority和attempt_count)。

这使运行器知晓是否存在待处理任务。下一步是申领其中一项任务。因此我们再次调用peek，若返回任务ID，则尝试申领该特定任务。

以下是比项目中QueueStore实现更基础清晰的版本：

def claim_first_available(
    self, worker_id: str, attempts: int = 3
) -> models.Task | None:
    qs = models.Task.filter(
        worker_id="", 
        status=TaskResultStatus.READY,
    )
    for _ in range(attempts):
        task_id = self.peek()
        if not task_id:
            return None
        count = qs.filter(pk=task_id).update(
            worker_id=self.id_,
            status=TaskResultStatus.RUNNING,
        )
        if count:
            return models.Task.objects.get(pk=task_id)
    return None

若count为零，则申领失败；否则从数据库获取任务并开始处理。

循环机制的引入是因为当前状态是尝试申领peek识别出的任务后跳转至此，显然该任务已被其他工作者抢先处理。既然如此，不妨充分利用当前状态，尝试从队列中获取另一个任务。

处理任务

终于要进入真正执行操作的环节了！

后端process_task方法的流程：

1. 创建Attempt对象并构造当前TaskResult。
2. 执行任务，捕获所有继承自BaseException的异常，或在执行顺利时返回任务的return_value。
3. 根据执行结果更新Task模型、Attempt对象及TaskResult：成功时记录最终执行细节，失败时记录失败原因。
4. 若执行失败：检查任务是否可重试。

再次说明：若需深入细节，请查阅代码库。

至此结束

当然，这个演示项目省略了所有需要深入思考的部分，比如工作进程的信号机制或数据库事务逻辑。这并非意味着无法实现——恰恰相反。只是本文并非探讨这些内容。

Django 引入这项功能后，必然会催生针对现有任务队列的新库或适配器。我们很快就会看到有人抱怨 django.tasks 功能不够完善。

因为如果你正在使用任务队列的高级功能，那么 django.tasks 可能无法满足你的需求。

复杂编排

某些任务队列库(如Celery)支持任务组合功能：可将一个任务的结果作为另一个任务的输入，为列表中的每个项排队处理任务等。

至此应已明确，支持此类编排并非django.tasks的目标。对此我完全不介意。要创建统一的API支持此功能根本不可行。我曾深陷那些声称支持此功能的库带来的各种问题。

重试机制

如前所述，当前无法自动重试失败任务，除非后端自行承担重任——就像我们的实现那样。

根据后端特性，这通常可自行处理。例如使用装饰器：

def retry(func):
    @functools.wraps(func)
    def wrapper(context: TaskContext, *args, **kwargs):
        try:
            return func(context, *args, **kwargs)
        except BaseException as e:
            result = context.task_result
            backoff = math.pow(2, result.attempts)
            run_after = datetime.now(tz=UTC) + timedelta(seconds=backoff)
            result.task.using(run_after=run_after).enqueue(*args, **kwargs)
            raise e
    return wrapper


@task(takes_context=True)
@retry
def send_email(context: TaskContext, to: str, subject: str, body: str):
    # Do your thing 
    ...

真正的任务执行机制

确实如此。但参考实现确实提供了真正的任务执行器。请耐心等待，或者更好：加入贡献行列！

没有完美解决方案

我认为django.tasks很快就能覆盖至少80%的常见用例。是的，它的API简单且有限，但对我而言这更像是优势而非缺陷。我认为这已是最接近标准化方案的实现。

本文文字及图片出自 A first look at Django's new background tasks