Huey 介绍：Django 的 Celery 替代品

背景

所以今天，我将讨论一种名为Huey的 Celery 替代品，它比Celery更容易设置，而且体积也比 Celery 小得多。

我决定尝试 Huey 的原因是我在使用 Celery 执行一些常见任务时有时会遇到一些问题，因为文档不是太好。

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

先决条件

我们将安装以下软件包：

• redis
• Django
• 休伊
• 请求（可选，演示所需）

GitHub 仓库

以下博客附带一个 GitHub repo，您可以使用它来测试我们将要创建的演示项目。

单击此处查看 repo。

项目设置

创建项目目录

打开终端，输入以下内容来创建目录；您可以跳过此步骤并从文件资源管理器本身执行此操作。

 mkdir huey_demo

虚拟环境

• 让我们首先创建一个虚拟环境来安装我们的项目依赖项：

   python -m venv venv

  
  

• 激活虚拟环境（Linux）：

   source venv/bin/activate

安装依赖项

在终端中输入以下命令来安装所有依赖项：

 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 startapp demo

  
  

• 将创建的应用程序包含在项目settings.py中，并进行以下更改：

   INSTALLED_APPS = [ # Existing Apps "demo.apps.DemoConfig", # <== Add this line ]

  
  

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

   DEBUG=False

  我们将 Debug 设置为 False，以便我们能够看到 Huey 在生产中的运行情况，稍后会详细介绍。

项目概况

现在我们已经完成了项目设置，现在是时候向您介绍我们今天要构建的内容了。

我们将每天从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文件中：

    from django.db import models class Word(models.Model): word = models.CharField(max_length=200) part_of_speech = models.CharField(max_length=100) definition = models.TextField() example = models.TextField() def __str__(self): return self.word

   
   

2. 进行迁移：

    python manage.py makemigrations demo

   
   

3. 应用迁移：

    python manage.py migrate demo

任务定义

在演示应用程序目录中创建一个名为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. 休伊装饰者

   from huey.contrib.djhuey import db_periodic_task

   这是 Huey 提供的一个装饰器，用于注册涉及与数据库交互的定期任务，任务完成后该装饰器会自动关闭数据库连接，更多详细信息可以参考这里。

   
   

2. Crontab 计划

   @db_periodic_task(crontab(hour="18", minute="00"))

   
   

   我们将参数crontab(hour="18", minute="00")传递给我们的定期任务装饰器，这告诉 Huey 每天下午 6 点运行我们的任务。您可以利用此网站创建您的 crontab 计划，我每次都使用它。

   
   

3. Wordnik API 密钥

   from django.conf import settings # Usage ## settings.WORDNIK_API_KEY

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

   
   

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

   
   

4. Wordnik API 调用

   r = requests.get( f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}")

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

   
   

5. 将单词存储在数据库中

   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"] )

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

   
   

6. Wordnik API 响应

   以下是 Wordnik API 对 Word of the Day 端点的响应。为简洁起见，响应中的一些不相关部分已被截断。

    { "word": "stolon", "definitions": [ { "source": "ahd-5", "text": "A long thin stem that usually grows horizontally along the ground and produces roots and shoots at widely spaced nodes, as in a strawberry plant.", "note": null, "partOfSpeech": "noun" }, // More definitions here... ], "publishDate": "2022-05-08T03:00:00.000Z", "examples": [ { "title": "4.1 Nursery establishment", "text": "A stolon is a stem that grows along the ground, producing at its nodes new plants with roots and upright stems.", // Additional data here... }, // More examples here... ], // Additional fields here... }

正在运行的休伊工人

您可以在终端中输入以下命令来启动 Huey 工作器：

 python manage.py run_huey

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

• -v, --verbose - 详细日志记录（包括 DEBUG 级别）
• -q, --quiet – 最少日志记录
• -S, --simple - 简单日志格式（“时间消息”）

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

您还可以用 Huey 做什么？

任务装饰器

Huey 带有多个任务装饰器，具体取决于您在任务中执行的操作。

下面我将简单解释一下它们的作用。

以下是所有装饰器的导入语句：

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

• task ：常规任务。
• periodic_task ：当您想要根据计划定期运行任务时。
• db_task ：当您想要在任务中执行 DB 操作时。
• db_periodic_task ：当您想要在周期性任务中执行 DB 操作时。

Crontab 示例

让我向您展示一些如何使用 crontab 来安排任务的示例。

• crontab(minute='*/3')将安排任务每三分钟运行一次。
• crontab(hour='*/3', minute='5')将创建一个每三个小时后五分钟运行一次的任务。
• crontab(minute='00', hour='10', month='*/2', day_of_week='*/5')将创建一个任务，该任务在每个月的第 5 天上午 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=True ，Huey 将要求您设置 Redis 并使用run_huey命令运行 Huey Worker。

芹菜 VS 休伊

与 Celery 相比，Huey 的一些观察结果如下：

• 与 Celery 相比，依赖项占用空间更小。Celery 会同时安装 kombu 和 billiards。而 Huey 没有任何依赖项。

• 较少的服务需要为周期性任务运行，Celery 需要运行 beat 服务和 worker 服务来处理周期性任务，而我们只需要使用run_huey命令运行一个服务。

参考

1. 休伊文档
2. Wordnik 接口
3. 相关 Github Repo