Django 中的异步任务与 Django Q

学习如何将 Django Q（任务队列）与 Redis 代理结合使用，以卸载 Django 应用程序中长时间运行的任务。

要求

要跟着做，你需要：

• 如果你想使用 Heroku 的 Redis 插件，你需要一个 Heroku 账户。
• 已在您的系统上安装 Heroku CLI
• 较新版本的 Python，理想情况下是 3.6 或 3.7 版本。
• Git

在 Heroku 上部署是可选的，如果您本地已经有 Redis 实例，也可以使用您自己的 Redis 实例。

项目设置

现在让我们开始吧！首先，我们将创建一个新的 Python 虚拟环境，并安装 Django：

mkdir django-q-django && cd $_
python3 -m venv venv
source venv/bin/activate
pip install django

接下来，我们将从模板创建一个新的 Django 项目：

django-admin startproject \
    --template https://github.com/valentinogagliardi/ponee/archive/master.zip \
    --name=Procfile \
    --extension=py,example django_q_django .

如果你好奇我在这里做什么，这是我自己的一个模板。资源部分有一个链接，里面有创建你自己的 Django 项目模板的教程。

现在让我们用pip安装依赖项：

pip install -r ./requirements/dev.txt

我们还需要为项目提供一些环境变量：

mv .env.example .env

最后，我们将运行 Django 迁移：

python manage.py makemigrations
python manage.py migrate

此时你应该能够运行开发服务器了：

python manage.py runserver

在深入了解 Django Q 之前，让我们先来看看它旨在解决什么问题。

Django 中的异步任务以及 Django Q：同步代码的问题

Python 和 Django 的主要问题在于它们是同步的。这本身并不是坏事，而且有很多方法可以解决这个问题。

Django 所基于的 Python 语言本质上是单线程的。单线程意味着语言解释器只能按顺序执行你的代码。

实际意义在于，如果一个或多个操作耗时过长，Django 应用程序中的任何视图都可能卡住。

为了演示这个概念，让我们在项目中创建一个新的 Django 应用程序：

django-admin startapp demo_app

在这个应用中，我们将定义一个视图，该视图返回一个简单的JSON响应：

# demo_app/views.py

from django.http import JsonResponse

def index(request):
    json_payload = {
        "message": "Hello world!"
    }
    return JsonResponse(json_payload)

我们再来创建相应的网址：

# demo_app/urls.py

from django.urls import path
from .views import index

urlpatterns = [
    path("demo-app/", index)
]

别忘了把新应用的网址设置进去：

# django_q_django/urls.py

from django.contrib import admin
from django.urls import path, include
from .settings.base import ADMIN_URL

urlpatterns = [
    path(f'{ADMIN_URL}/', admin.site.urls),
    # the new url
    path("", include("demo_app.urls"))
]

最后激活应用程序：

# django_q_django/settings/base.py

INSTALLED_APPS = [
    # omitted for brevity
    'demo_app.apps.DemoAppConfig'
]

现在，为了模拟视图中的阻塞事件，我们将使用Python 标准库中的 time 模块里的 sleep 函数：

from django.http import JsonResponse
from time import sleep

def index(request):
    json_payload = {
        "message": "Hello world!"
    }
    sleep(10)
    return JsonResponse(json_payload)

运行开发服务器，访问http://127.0.0.1:8000/demo-app/，可以看到视图卡顿 10 秒钟后才返回给用户。

现在，这是人为设置的延迟，但在实际应用中，阻塞可能由多种原因造成：

• I/O 操作耗时过长
• 网络延迟
• 与文件系统的交互

即使这是一个人为设计的例子，你也可以看出为什么在 Web 应用程序中卸载长时间运行的任务至关重要。

Django Q 的诞生正是为了实现这个目标。在接下来的章节中，我们将最终实现它。

等等，异步 Django 怎么样？

我们现在得到的是一个 Django 项目、一个 Django 应用程序，以及一个卡住 10 秒钟的视图！

并非Django和Python不具备可扩展性。有很多方法可以克服单线程的限制。

Python 有 asyncio。而 Django 直到最近才转向 async，其实现还处于起步阶段，目前还不支持异步视图。

未来情况会有所变化，我建议大家关注Andrew Godwin，因为他是 Django 异步开发的主要负责人。

即使Django 完全实现异步，对第三方队列的需求也不会很快消失。掌握这项技能仍然很有用。

准备 Heroku 应用和 Redis 实例

在本节中，我们将准备 Heroku 项目。我在这里使用 Heroku 是因为你以后可能需要部署到生产环境，而且他们还免费提供 Redis 插件。

如果你是 Redis 的新手，它是一个内存数据库，可以用作缓存和消息代理。

消息代理服务器或多或少就像一个邮局信箱：它接收消息，将消息保存在队列中，城市各地的人们可以稍后检索这些消息。

如果您对 Django Q 如何使用 brokers 感兴趣，请查看此页面。

仍在项目文件夹中，初始化一个 Git 仓库：

git init

然后创建一个新的 Heroku 应用。我将添加两个插件：

• heroku-postgresql 比默认的 sqlite 更稳定，更适合生产环境。
• heroku-redis将为我们提供 Redis 实例。

如果你还没有 Heroku CLI 和 Heroku 帐户，请创建一个，安装 CLI，稍后再回来。

否则，请跟随我的步骤创建应用程序：

heroku create --addons=heroku-postgresql,heroku-redis

完成后，请等待 Heroku 几分钟，然后运行：

heroku config:get REDIS_URL

此命令将显示REDIS_URL，这是一个包含 Redis 实例凭据的环境变量。

记下这一点，然后前往下一节！

Django Q 中的异步任务：安装和运行 Django Q

让我们安装 Django Q 和 Redis 客户端库（Django Q 的 Redis 代理需要客户端）：

pip install django-q redis

完成后，在已安装的应用列表中激活 Django Q：

INSTALLED_APPS = [
    # omit
    # add Django Q
    'django_q'
]

现在显示 Redis Heroku 凭据：

heroku config:get REDIS_URL

你应该会看到类似这样的字符串：

redis://h:p948710311f252a334c3b21cabe0bd63f943f68f0824cd41932781e7793c785bf@ec2-52-18-11-1.eu-west-1.compute.amazonaws.com:9059

@符号之前就是密码：

p948710311f252a334c3b21cabe0bd63f943f68f0824cd41932781e7793c785bf

@符号后面是主机名：

ec2-52-18-11-1.eu-west-1.compute.amazonaws.com

端口号是9059。请注意，您的凭据与我的不同，请勿使用我的！

（毋庸置疑，当你读到这篇文章时，这些​​资质早已失效。）

现在在django_q_django/settings/base.py中配置 Django Q。请将主机、端口和密码填写为您的凭据：

Q_CLUSTER = {
    'name': 'django_q_django',
    'workers': 8,
    'recycle': 500,
    'timeout': 60,
    'compress': True,
    'save_limit': 250,
    'queue_limit': 500,
    'cpu_affinity': 1,
    'label': 'Django Q',
    'redis': {
        'host': 'ec2-52-18-11-1.eu-west-1.compute.amazonaws.com',
        'port': 9059,
        'password': 'p948710311f252a334c3b21cabe0bd63f943f68f0824cd41932781e7793c785bf',
        'db': 0, }
}

你可能想知道为什么我没有直接使用 REDIS_URL。原因是 Django Q 要求凭据以字典形式存储。

我还没时间检查是否是 Python Redis 客户端造成了这种限制，也许将来我会为两者都写一个补丁。这是 Django Q 的一个局限性。 希望我有时间提交一个 PR。我提交了一个拉取请求，该请求已被合并，现在您可以使用 Redis URL 了：

Q_CLUSTER = {
    'name': 'django_q_django',
    # omitted for brevity  
    'label': 'Django Q',
    'redis': 'redis://h:asdfqwer1234asdf@ec2-111-1-1-1.compute-1.amazonaws.com:111'
}

（在生产环境中运行项目时，您可能需要切换到使用环境变量。请参阅基本配置以了解如何使用环境变量。）

完成后运行迁移（Django Q 需要在数据库中创建表）：

python manage.py migrate

至此，您已准备好运行 Django Q 集群：

python manage.py qcluster

如果一切顺利，你应该会看到这个：

做得好！下一节我们将创建第一个异步任务。

什么是 Django Q 集群？点击这里了解详情。

Django 中的异步任务（使用 Django Q: async_task）

值得快速回顾一下我们目前为止讨论的内容：

• 我们创建了一个 Django 项目
• 我们创建了一个 Django 应用程序
• 我们安装了 Django Q 和 Redis 客户端
• 我们创建了一个 Heroku 项目和一个 Redis 实例。
• 最后我们配置了 Django Q。

为了测试 Django Q 能否连接到 Redis，我启动了以下命令：

python manage.py qcluster

项目已经准备就绪，让我们终于可以看到Django Q 的实际应用示例了。还记得你的视图吗？

# demo_app/views.py

from django.http import JsonResponse
from time import sleep

def index(request):
    json_payload = {
        "message": "Hello world!"
    }
    sleep(10)
    return JsonResponse(json_payload)

删除时间导入，并在demo_app/services.py中创建一个新文件（该文件的名称完全由您决定）。

在这个新模块中，我们将定义一个函数，sleep_and_print：

# demo_app/services.py

from time import sleep

def sleep_and_print(secs):
    sleep(secs)
    print("Task ran!")

在视图中，我们将借用Django Q 的async_task ：

from django.http import JsonResponse
from django_q.tasks import async_task


def index(request):
    json_payload = {
        "message": "hello world!"
    }
    """
    TODO
    """
    return JsonResponse(json_payload)

async_task是你在 Django Q 中会用到的主要函数。它至少接受一个参数，即你要加入队列的函数模块：

# example

async_task("demo_app.services.sleep_and_print")

第二组参数则是函数应该接收的任何参数。例如，`sleep_and_print` 函数在我们的示例中接受一个参数，即打印前等待的秒数。这意味着对于 `async_task`：

# example

async_task("demo_app.services.sleep_and_print", 10)

这样就足以将任务加入队列了。现在让我们把视图与 async_task结合起来。

Django 中的异步任务：使用 Django Q 将你的第一个任务加入队列

回到我们之前的视图，导入 async_task 后，在 return 语句之后立即调用它：

from django.http import JsonResponse
from django_q.tasks import async_task


def index(request):
    json_payload = {"message": "hello world!"}
    # enqueue the task
    async_task("demo_app.services.sleep_and_print", 10)
    #
    return JsonResponse(json_payload)

现在运行集群：

python manage.py qcluster

运行 Django 服务器：

python manage.py runserver

最后，通过http://127.0.0.1:8000/demo-app/或终端调用您的视图：

curl http://127.0.0.1:8000/demo-app/

现在你应该注意到几件事。Django 开发服务器应该会记录以下日志：

13:55:42 [Q] INFO Enqueued 1

Django Q 集群应该会记录类似这样的信息：

13:55:42 [Q] INFO Process-1:1 processing [juliet-mountain-august-alaska]

之后你应该会看到：

Task ran!

这是我的终端：

事情的经过是这样的：

1. Django视图立即响应了请求。
2. Django Q 将任务（仅作引用）保存到了 Redis 中
3. Django Q 运行了该任务

有了这种“架构”，视野不再局限于原地。妙哉！

思考一下这种模式的应用场景。您可以：

• 安全地与 I/O 进行交互
• 在后台处理数据
• 安全地将 API 调用从视图中移出

还有更多。

Django 中的异步任务（使用 Django Q：下一步是什么？）

除了 `async_task` 之外，Django Q 还支持定时任务。一个实际应用场景是每隔 X 天执行一次 X 操作，类似于 cron 任务。请查阅文档了解更多信息。

Django Q 除了 Redis 之外，还支持其他消息代理。再次强调，文档是你的好帮手。

如果您不需要 Redis 以外的其他代理，那么django-rq可能是 Django Q 的一个轻量级替代方案。

Django 中的异步任务与 Django Q：为什么不使用 Celery？

有趣的是：芹菜是我一个朋友发明的。我们高中是同学。虽然我对芹菜本身了解不多，但我总是听到很多人抱怨它。

看看这个，或许能有更全面的了解。

感谢阅读，敬请期待！

资源

• Django Q
• Redis
• 什么是消息代理？
• 后台作业和队列
• 如何从模板创建 Django 项目

原文发表于我的博客