使用 Django 的第一步

在 Django 中使用芹菜

笔记

以前版本的 Celery 需要一个单独的库来与 Django 一起工作，但从 3.1 开始，情况不再如此。 Django 现在支持开箱即用，因此本文档仅包含集成 Celery 和 Django 的基本方法。您将使用与非 Django 用户相同的 API，因此建议您先阅读使用 Celery 的第一步教程，然后再返回本教程。有了工作示例后，您可以继续阅读后续步骤指南。

笔记

Celery 5.0.x 支持 Django 1.11 LTS 或更新版本。对于 Django 1.11 之前的版本，请使用 Celery 4.4.x。

要在 Django 项目中使用 Celery，您必须首先定义 Celery 库的一个实例（称为“应用程序”）

如果你有一个现代的 Django 项目布局，比如：

- proj/
  - manage.py
  - proj/
    - __init__.py
    - settings.py
    - urls.py

那么推荐的方法是创建一个新的 proj/proj/celery.py 模块来定义 Celery 实例：

档案: 项目/项目/芹菜.py

import os

from celery import Celery

# Set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

app = Celery('proj')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django apps.
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

然后你需要在你的 proj/proj/__init__.py 模块中导入这个应用程序。这确保在 Django 启动时加载应用程序，以便 @shared_task 装饰器（稍后提到）将使用它：

proj/proj/__init__.py：

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ('celery_app',)

请注意，此示例项目布局适用于较大的项目，对于简单的项目，您可以使用单个包含的模块来定义应用程序和任务，如芹菜的第一步教程。

让我们分解第一个模块中发生的事情，首先，我们为 celery 命令行程序设置默认的 DJANGO_SETTINGS_MODULE 环境变量：

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

您不需要这一行，但它可以避免您总是将设置模块传递给 celery 程序。它必须始终在创建应用程序实例之前出现，就像我们接下来要做的一样：

app = Celery('proj')

这是我们的库实例，您可以有很多实例，但在使用 Django 时可能没有理由这样做。

我们还添加了 Django 设置模块作为 Celery 的配置源。这意味着您不必使用多个配置文件，而是直接从 Django 设置中配置 Celery；但如果需要，您也可以将它们分开。

app.config_from_object('django.conf:settings', namespace='CELERY')

大写命名空间意味着所有 Celery 配置选项必须以大写而不是小写指定，并以 CELERY_ 开头，例如 :setting:`task_always_eager`[ X208X] 设置变为 CELERY_TASK_ALWAYS_EAGER，:setting:`broker_url` 设置变为 CELERY_BROKER_URL。这也适用于 worker 设置，例如，:setting:`worker_concurrency` 设置变为 CELERY_WORKER_CONCURRENCY。

例如，Django 项目的配置文件可能包括：

设置.py

...

# Celery Configuration Options
CELERY_TIMEZONE = "Australia/Tasmania"
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60

您可以直接传递设置对象，但使用字符串更好，因为这样工作人员不必序列化对象。 CELERY_ 命名空间也是可选的，但建议使用（以防止与其他 Django 设置重叠）。

接下来，可重用应用程序的常见做法是在单独的 tasks.py 模块中定义所有任务，Celery 确实有一种方法可以自动发现这些模块：

app.autodiscover_tasks()

Celery 将按照 tasks.py 约定自动从所有已安装的应用程序中发现任务：

- app1/
    - tasks.py
    - models.py
- app2/
    - tasks.py
    - models.py

这样您就不必手动将各个模块添加到：设置：`CELERY_IMPORTS ` 环境。

最后，debug_task 示例是一个转储自己请求信息的任务。这是使用 Celery 3.1 中引入的新 bind=True 任务选项来轻松引用当前任务实例。

使用 @shared_task 装饰器

您编写的任务可能会存在于可重用应用程序中，而可重用应用程序不能依赖于项目本身，因此您也无法直接导入您的应用程序实例。

@shared_task 装饰器使您无需任何具体的应用程序实例即可创建任务：

demoapp/tasks.py：

# Create your tasks here

from demoapp.models import Widget

from celery import shared_task


@shared_task
def add(x, y):
    return x + y


@shared_task
def mul(x, y):
    return x * y


@shared_task
def xsum(numbers):
    return sum(numbers)


@shared_task
def count_widgets():
    return Widget.objects.count()


@shared_task
def rename_widget(widget_id, name):
    w = Widget.objects.get(id=widget_id)
    w.name = name
    w.save()

也可以看看

您可以在以下位置找到 Django 示例项目的完整源代码：https://github.com/celery/celery/tree/master/examples/django/

扩展

django-celery-results - 使用 Django ORM/Cache 作为结果后端

:pypi:`django-celery-results` 扩展使用 Django ORM 或 Django 缓存框架提供结果后端。

要在您的项目中使用它，您需要按照以下步骤操作：

安装 :pypi:`django-celery-results` 库：
$ pip install django-celery-results
在 Django 项目的 settings.py 中将 django_celery_results 添加到 INSTALLED_APPS：
```
INSTALLED_APPS = (
    ...,
    'django_celery_results',
)
```
请注意，模块名称中没有破折号，只有下划线。
通过执行数据库迁移来创建 Celery 数据库表：
$ python manage.py migrate django_celery_results
配置 Celery 以使用 :pypi:`django-celery-results` 后端。
假设您还使用 Django 的 settings.py 来配置 Celery，请添加以下设置：
CELERY_RESULT_BACKEND = 'django-db'
对于缓存后端，您可以使用：
CELERY_RESULT_BACKEND = 'django-cache'
我们也可以使用 django 中 CACHES 设置中定义的缓存。
# celery setting. CELERY_CACHE_BACKEND = 'default' # django setting. CACHES = { 'default': { 'BACKEND': 'django.core.cache.backends.db.DatabaseCache', 'LOCATION': 'my_cache_table', } }
有关其他配置选项，请查看任务结果后端设置参考。

django-celery-beat - 具有管理界面的数据库支持的周期性任务。

有关更多信息，请参阅使用自定义调度程序类。

启动工作进程

在生产环境中，您将希望在后台运行 worker 作为守护进程 - 参见 Daemonization - 但对于测试和开发，能够使用 启动一个 worker 实例很有用celery worker 管理命令，就像你使用 Django 的 manage.py runserver：

$ celery -A proj worker -l INFO

有关可用命令行选项的完整列表，请使用 help 命令：

$ celery help

从这里去哪里

如果您想了解更多信息，请继续阅读后续步骤教程，之后您可以学习用户指南。

Django 的第一步 — Python 文档

目录