I write code.
Walkthroughs, tutorials, guides, and tips. This story will teach you how to do something new or how to do something better.
所以今天,我将讨论一种名为Huey的 Celery 替代品,它比Celery更容易设置,而且体积也比 Celery 小得多。
我决定尝试 Huey 的原因是我在使用 Celery 执行一些常见任务时有时会遇到一些问题,因为文档不是太好。
对于那些不知道 Celery 是什么或以前没有使用过的人来说,Huey 是一个异步任务队列,允许您在后台执行计划任务或长时间运行的任务。
我们将安装以下软件包:
以下博客附带一个 GitHub 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 模型。
您可以按照本指南获取 API 密钥。
我们需要将 Huey 添加到我们项目的已安装应用程序中,因此在settings.py
文件中进行以下更改:
INSTALLED_APPS = [ # Existing apps 'huey.contrib.djhuey', # <== Add this line ]
我们需要为 Huey 安装 Redis,以便像以前使用 Celery 一样在其中存储有关排队任务的信息。您可以参考以下链接根据您的特定操作系统安装 Redis。
如果您熟悉使用 Docker,则可以使用以下命令:
docker run --name redis_huey -p 6379:6379 -d redis
默认情况下,Huey 将尝试连接到在localhost:6379
上运行的 Redis 服务器。如果不存在,则会引发错误。
将以下代码添加到您的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
在演示应用程序目录中创建一个名为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"
这段代码块可能需要理解很多内容,我们来逐一看一下:
休伊装饰者
from huey.contrib.djhuey import db_periodic_task
这是 Huey 提供的一个装饰器,用于注册涉及与数据库交互的定期任务,任务完成后该装饰器会自动关闭数据库连接,更多详细信息可以参考这里。
Crontab 计划
@db_periodic_task(crontab(hour="18", minute="00"))
我们将参数crontab(hour="18", minute="00")
传递给我们的定期任务装饰器,这告诉 Huey 每天下午 6 点运行我们的任务。您可以利用此网站创建您的 crontab 计划,我每次都使用它。
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 响应后,我们将单词定义存储在数据库中。我们在这里使用get_or_create
方法而不是create
方法,这样如果 Wordnik API 重复了该单词,我们就不会在数据库中创建同一个单词的多个副本。
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 带有多个任务装饰器,具体取决于您在任务中执行的操作。
下面我将简单解释一下它们的作用。
以下是所有装饰器的导入语句:
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(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。
与 Celery 相比,Huey 的一些观察结果如下:
run_huey
命令运行一个服务。
Huey 介绍:Django 的 Celery 替代品 | HackerNoon