背景 所以今天，我将讨论一种名为 的 Celery 替代品，它比 更容易设置，而且体积也比 Celery 小得多。 Huey 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 将尝试连接到在 上运行的 Redis 服务器。如果不存在，则会引发错误。 localhost:6379 模型定义 将以下代码添加到您的 文件中： 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 进行迁移：   python manage.py makemigrations demo 应用迁移：   python manage.py migrate demo 任务定义 在演示应用程序目录中创建一个名为 的文件。我们将文件命名为 的原因是为了帮助Huey自动发现我们注册的应用程序中存在的任务，如果我们将文件命名为其他名称，我们将不得不手动注册我们的任务。如果您想了解更多信息，可以 查看Huey文档。 tasks.py tasks.py 在此处 在编写任务定义之前，我们需要安装一个额外的依赖项  。在终端中输入以下内容来安装它： 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" 这段代码块可能需要理解很多内容，我们来逐一看一下： 休伊装饰者 from huey.contrib.djhuey import db_periodic_task 这是 Huey 提供的一个装饰器，用于注册涉及与数据库交互的定期任务，任务完成后该装饰器会自动关闭数据库连接，更多详细信息可以参考 这里。   Crontab 计划 @db_periodic_task(crontab(hour="18", minute="00")) 我们将参数 传递给我们的定期任务装饰器，这告诉 Huey 每天下午 6 点运行我们的任务。您可以利用 创建您的 crontab 计划，我每次都使用它。 crontab(hour="18", minute="00") 此网站   Wordnik API 密钥 from django.conf import settings # Usage ## settings.WORDNIK_API_KEY   是从我们的项目设置导入任何数据的标准方法，在我们为不同环境设置多个设置文件的情况下，这种方法很有用，这样它就知道从哪个文件中选择，而我们不必担心。它从 环境变量中找出我们正在使用的设置文件。但您不必担心这些细节。 from django.conf import settings DJANGO_SETTINGS_MODULE 然后我们在 Wordnik API 调用中使用该密钥。   Wordnik API 调用 r = requests.get( f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}") 在这里，我们利用请求模块向 Wordnik API 发出 GET 请求，同时传递我们的 API 密钥进行身份验证。 将单词存储在数据库中 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 响应后，我们将单词定义存储在数据库中。我们在这里使用 方法而不是 方法，这样如果 Wordnik API 重复了该单词，我们就不会在数据库中创建同一个单词的多个副本。 get_or_create create   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 您可以将多个标志传递给上述命令，这将改变记录到控制台的内容，例如：    - 详细日志记录（包括 DEBUG 级别） -v, --verbose    – 最少日志记录 -q, --quiet  - 简单日志格式（“时间消息”） -S, --simple 要查看其他各种日志记录选项，请查看 文档。 此处的 您还可以用 Huey 做什么？ 任务装饰器 Huey 带有多个任务装饰器，具体取决于您在任务中执行的操作。 下面我将简单解释一下它们的作用。 以下是所有装饰器的导入语句：   from huey.contrib.djhuey import task, periodic_task, db_task, db_periodic_task    ：常规任务。 task    ：当您想要根据计划定期运行任务时。 periodic_task    ：当您想要在任务中执行 DB 操作时。 db_task    ：当您想要在周期性任务中执行 DB 操作时。 db_periodic_task  Crontab 示例 让我向您展示一些如何使用 crontab 来安排任务的示例。   将安排任务每三分钟运行一次。 crontab(minute='*/3')   将创建一个每三个小时后五分钟运行一次的任务。 crontab(hour='*/3', minute='5')   将创建一个任务，该任务在每个月的第 5 天上午 10:00 运行。 crontab(minute='00', hour='10', month='*/2', day_of_week='*/5') 安排任务 例如，在 中定义了以下任务： tasks.py   from huey.contrib.djhuey import task @task() def count(): for i in range(10): print(i) 现在，如果您想调用此任务但希望它在 5 秒后运行，您可以执行以下操作：   count.schedule(delay=5)   参数以秒为单位取值，因此如果您希望它在 5 分钟后执行，请指定 300 秒。 delay 重试失败的任务 假设您将以下逻辑添加到我们现有的任务中：   @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 更加容易。因此，只要您的 文件中存在  ，任务就会像常规函数调用一样同步执行。这样做的目的是避免在开发或运行测试时同时运行 Redis 和额外的消费者进程。您可以 阅读更多相关信息。 settings.py DEBUG=True 在此处 为此，我们需要在 中添加以下行： settings.py   HUEY = {} 但是，如果您想覆盖此行为，您可以添加以下 Huey 配置：   HUEY = { "immediate": False } 如果您在 中具有上述配置，同时具有  ，Huey 将要求您设置 Redis 并使用 命令运行 Huey Worker。 settings.py DEBUG=True run_huey 芹菜 VS 休伊 与 Celery 相比，Huey 的一些观察结果如下： 与 Celery 相比，依赖项占用空间更小。Celery 会同时安装 kombu 和 billiards。而 Huey 没有任何依赖项。 较少的服务需要为周期性任务运行，Celery 需要运行 beat 服务和 worker 服务来处理周期性任务，而我们只需要使用 命令运行一个服务。 run_huey 参考 休伊文档 Wordnik 接口 相关 Github Repo

Walkthroughs, tutorials, guides, and tips. This story will teach you how to do something new or how to do something better.

Portfolio

Read My Stories

該音頻是用故事的原始語言製作的！

Huey 介绍：Django 的 Celery 替代品

About Author

註釋

標籤

这篇文章刊登在

Related Stories

看不见的层面：为什么用户访谈是不可替代的资产

扬帆起航：利用数据湖开发生产级 RAG 应用程序

点击赚钱：Telegram 可能会在 Solana 之前吸引下一个 100 亿加密用户

成功云迁移的完整指南：策略和最佳实践

看不见的层面：为什么用户访谈是不可替代的资产

扬帆起航：利用数据湖开发生产级 RAG 应用程序

点击赚钱：Telegram 可能会在 Solana 之前吸引下一个 100 亿加密用户

成功云迁移的完整指南：策略和最佳实践

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps