背景

所以今天我将讨论一种名为Huey的 Celery 替代品,它的设置比Celery更容易,并且与 Celery 相比尺寸要小得多。

我决定尝试 Huey 的原因是因为有时我在 Celery 上遇到了一些问题,因为文档不是很好。

对于那些不知道 Celery 是什么或以前没有使用过它的人来说,Huey 是一个异步任务队列,它允许您在后台执行计划任务或长时间运行的任务。

先决条件

我们将安装以下软件包:

• redis

• django

• 休伊

• 个请求(可选,演示需要)

Github 回购

以下博客附带一个 github 存储库,您可以使用它来测试我们将创建的演示项目。

点击这里查看回购。

项目设置

创建项目目录

打开终端并键入以下内容以创建目录,您可以跳过此步骤并从文件资源管理器本身进行。

mkdir huey_demo

进入全屏模式 退出全屏模式

虚拟环境

• 让我们先创建一个 virtualenv 来安装我们的项目依赖项:

python -m venv venv

• 激活 virtualenv (Linux):

源 venv/bin/激活

安装依赖

在终端中键入以下命令以安装所有依赖项:

pip install Django==4.0.4 redis==4.2.2 huey==2.4.3

进入全屏模式 退出全屏模式

在撰写本文时,这些是我测试此设置的版本,请留意 Github Repo 以了解未来最新版本的任何更新。

创建项目

• 通过在终端中键入以下命令来创建 django 项目:

django-admin startproject django_huey_demo

• 将目录更改为 Django 项目目录:

cd django_huey_demo

• 在我们的项目下创建应用:

python manage.py 启动应用演示

• 将创建的应用程序包含到项目settings.py中,进行以下更改:

安装应用程序 u003d [
# 现有应用
"demo.apps.DemoConfig", # <u003du003d 添加这一行
]

• 在settings.py中设置调试模式为False:

调试u003d假

项目概况

现在我们已经完成了项目的设置,现在是让您了解我们今天将要构建的内容的好时机。我们将每天从Wordnik API获取“每日词汇”。然后我们将单词、它的定义和一个句子中的单词示例存储到我们的数据库中。

我们将使用 Huey 设置一个定期任务,该任务将获取每日单词并将其存储。

为了存储单词,我们将创建一个相同的 Django 模型。

获取 Wordnik API 密钥

您可以按照本指南获取 API 密钥。

编码我们的项目

将 Huey 添加到我们的项目中

我们需要将 Huey 添加到我们项目的已安装应用程序中,因此在settings.py文件中进行以下更改:

INSTALLED_APPS = [
    # Existing apps
    'huey.contrib.djhuey', # <== Add this line
]

进入全屏模式 退出全屏模式

安装Redis

我们需要为 Huey 安装 Redis 以在其中存储有关排队任务的信息,就像我们过去对 Celery 所做的那样。你可以参考下面的链接根据你的具体操作系统安装redis。

如果您习惯使用 Docker,可以使用以下命令:

docker run --name redis_huey -p 6379:6379 -d redis

进入全屏模式 退出全屏模式

默认情况下,Huey 将尝试连接到运行在localhost:6379上的 Redis 服务器。如果它不存在,它将引发错误。

型号定义

1. 将以下代码添加到您的demo/models.py文件中:

从 django.db 导入模型

类字(模型。模型):
word u003d models.CharField(max_lengthu003d200)
part_of_speech u003d models.CharField(max_lengthu003d100)
定义 u003d 模型.TextField()
示例 u003d 模型.TextField()

def __str__(self):
返回self.word

2. 进行迁移:

python manage.py makemigrations 演示

3. 应用迁移:

python manage.py 迁移演示

任务定义

在演示应用目录中创建一个名为tasks.py的文件。我们将文件命名为tasks.py的原因是为了帮助 Huey 自动发现我们注册的应用程序中存在的任务,如果我们将文件命名为其他名称,我们将不得不手动注册我们的任务。如果您想了解更多信息,可以在此处查看 Huey 文档。

在我们编写任务定义之前,我们需要安装一个额外的依赖项requests。通过在终端中键入以下内容来安装它:

pip install requests==2.27.1

进入全屏模式 退出全屏模式

现在是代码:

import requests
from django.conf import settings
from huey import crontab
from huey.contrib.djhuey import db_periodic_task

from demo.models import Word

@db_periodic_task(crontab(hour="18", minute="00"))
def fetch_daily_word():
    r = requests.get(
        f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}")
    data = r.json()
    Word.objects.get_or_create(
        word=data["word"],
        part_of_speech=data["definitions"][0]["partOfSpeech"],
        definition=data["definitions"][0]["text"],
        example=data["examples"][0]["text"]
    )

进入全屏模式 退出全屏模式

在项目设置中添加以下行:

WORDNIK_API_KEY = "api-key-here"

进入全屏模式 退出全屏模式

这个代码块可能有很多需要吸收的东西,所以让我们一一回顾一下:

1. 休伊装饰师

从 huey.contrib.djhuey 导入 db_periodic_task

这是 Huey 提供的一个装饰器,用于注册涉及到数据库的周期性任务,该装饰器在任务完成时自动关闭数据库连接,更多详细信息,您可以参考这里。

2. Crontab 时间表

@db_periodic_task(crontab(houru003d"18", minuteu003d"00"))

我们将参数crontab(hour="18", minute="00")传递给我们的周期性任务装饰器,这告诉 Huey 在每天下午 6 点运行我们的任务。你可以利用这个网站来创建你的 crontab 时间表,我每次都用它。

3. Wordnik API 密钥

从 django.conf 导入设置

# 用法
## settings.WORDNIK_API_KEY

from django.conf import settings是从我们的项目设置中导入任何数据的标准方式,它在我们为不同环境设置了多个设置文件的情况下很有用,因此它会知道从哪个文件中选择而无需我们担心。它从DJANGO_SETTINGS_MODULE环境变量中找出我们正在使用的设置文件。但是您不必担心这些细节。

然后我们在 wordnik API 调用中使用密钥。

4. Wordnik API 调用

r u003d 请求.get(
f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_keyu003d{settings.WORDNIK_API_KEY}")

在这里,我们使用 requests 模块向 wordnik API 发出 GET 请求,同时传递我们的 API 密钥进行身份验证。

5. 在数据库中存储单词

数据 u003d r.json()
Word.objects.get_or_create(
字u003d数据[“字”],
part_of_speechu003d数据["定义"][0]["partOfSpeech"],
定义u003d数据[“定义”][0][“文本”],
示例u003d数据[“示例”][0][“文本”]
)

解析 API 响应后,我们将单词定义存储在数据库中。我们在这里使用get_or_create方法而不是create方法,这样如果 Wordnik API 重复该词,我们就不会在数据库中创建同一个词的多个副本。

6. Wordnik API 响应

这是 Word of the Day 端点的 Wordnik API 响应的样子。为简洁起见,响应中一些不相关的部分已被截断。

{
“词”:“匍匐茎”,
“定义”:[
{
“来源”:“ahd-5”,
"text": "一根细长的茎,通常沿地面水平生长,并在间距较宽的节点处产生根和芽,如草莓植物。",
“注意”:空,
“partOfSpeech”:“名词”
},
// 这里有更多定义...
],
“发布日期”:“2022-05-08T03:00:00.000Z”,
“例子”: [
{
"title": "4.1 托儿所建立",
"text": "匍匐茎是沿着地面生长的茎,在其节点处产生具有根和直立茎的新植物。",
// 这里的附加数据...
},
// 这里有更多例子...
],
// 这里的附加字段...
}

这是 Word of the Day 端点的 Wordnik API 响应的样子。为简洁起见,响应中一些不相关的部分已被截断。

运行休伊工人

您可以通过在终端中键入以下命令来启动 Huey 工作程序:

python manage.py run_huey

进入全屏模式 退出全屏模式

您可以将多个标志传递给上述命令,这将更改记录到控制台的内容,例如:

• -v, --verbose- 详细日志记录(包括调试级别)

• -q, --quiet- 最小日志记录

• -S, --simple- 简单的日志记录格式(“时间消息”)

要查看其他各种日志记录选项,请在此处查看文档。

休伊还能做什么?

任务装饰器

Huey 带有多个任务装饰器,具体取决于您在任务中执行的操作。我将在下面简要解释所有这些的作用。这是所有装饰器的导入语句:

from huey.contrib.djhuey import task, periodic_task, db_task, db_periodic_task

进入全屏模式 退出全屏模式

• task:常规任务

• periodic_task: 当您想根据计划定期运行任务时

• db_task:当你想在你的任务中执行数据库操作时

• db_periodic_task: 当你想在周期性任务中执行 db 操作时

crontab 示例

让我向您展示更多示例,说明如何使用 crontab 来安排任务。

• crontab(minute='*/3')将安排任务每三分钟运行一次

• crontab(hour='*/3', minute='5')将创建一个将在每三个小时后 5 分钟运行的任务。

• crontab(minute='00', hour='10', month='*/2', day_of_week='*/5')将创建一个任务,该任务将在每周的第 5 天,每 2 个月的上午 10:00 运行。

调度任务

例如,您在tasks.py中定义了以下任务:

from huey.contrib.djhuey import task

@task()
def count():
    for i in range(10):
        print(i)

进入全屏模式 退出全屏模式

现在您想调用此任务,但希望它在 5 秒后运行,您可以执行以下操作:

count.schedule(delay=5)

进入全屏模式 退出全屏模式

delay参数以秒为单位取值,因此如果您希望它在 5 分钟后执行,请指定 300 秒。

重试失败的任务

假设您将以下逻辑添加到我们现有的任务中:

@db_periodic_task(crontab(hour="18", minute="00"), retries=2)
def fetch_daily_word():
    r = requests.get(
        f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}")
    if r.status_code != 200:
        raise Exception("Unable to fetch data from Wordnik API") ## Add this logic
    else:
        data = r.json()
        Word.objects.get_or_create(
            word=data["word"],
            part_of_speech=data["definitions"][0]["partOfSpeech"],
            definition=data["definitions"][0]["text"],
            example=data["examples"][0]["text"]
        )

进入全屏模式 退出全屏模式

所以我们添加了检查响应状态码的逻辑,如果不是 200,它将重试该任务最多 2 次。但是这些重试将在两次尝试之间没有任何时间间隔的情况下发生。现在,如果您想延迟此任务的多次尝试,我们可以通过传递retry_delay参数来做到这一点,它接受以秒为单位的值。

@db_periodic_task(crontab(hour="18", minute="00"), retries=2, retry_delay=10)

进入全屏模式 退出全屏模式

这将导致多次尝试之间有 10 秒的延迟。

发展模式

Huey 带有一个默认设置,这使得在 Django 开发过程中使用 Huey 更容易。因此,只要settings.py文件中存在DEBUG=True,任务就会像常规函数调用一样同步执行。这样做的目的是避免在开发或运行测试时同时运行 Redis 和额外的消费者进程。您可以在此处阅读有关此的更多信息。

为此,我们需要在settings.py中添加以下行:

HUEY = {}

进入全屏模式 退出全屏模式

但是,如果您想覆盖此行为,则可以添加以下 Huey 配置:

HUEY = {
    "immediate": False
}

进入全屏模式 退出全屏模式

如果你有上述settings.py中提到的配置,同时拥有DEBUG=TrueHuey 将需要你设置 Redis 并使用run_huey命令运行 huey worker。

芹菜与休伊

与 Celery 相比,关于 Huey 的一些观察结果是:

• 与 Celery 相比,依赖项的占用空间更小。芹菜安装了昆布和台球。同时,huey 没有任何依赖关系。

• 周期性任务需要运行较小的服务,Celery 需要运行 beat 服务和工作服务来处理周期性任务,同时我们只需要使用run_huey命令运行一个服务。

参考文献

1.Huey Docs

2.Wordnik API

3.关联Github Repo