Huey
文章目录
前言
特性:
支持:
一、Huey是什么?
Huey是个Python任务队列,是个轻量级的替代品,没有多余的依赖关系。
二、使用步骤
1.安装
huey 可以使用pip从 PyPI 安装。
pip install huey
huey 在标准库之外没有依赖项,但是需要redis-py 才能利用 Redis 进行任务存储:
pip install redis
2.如何在我们的项目中进行配置
Huey带有与Django框架一起使用的特殊集成。该集成提供
支持的Django的版本是 xxxx
3.设置
要在Django中使用Huey,第一步是在项目中注册Huey的app
# settings.py
# ...
INSTALLED_APPS = (
# ...
'huey.contrib.djhuey', # 将此添加到列表中。
# ...
)
这是在Django中使用Huey集成所需要的最低需求,我们还可以在settings中继续做一些其他的配置以便方便我们的使用。
# settings.py
HUEY = {
'huey_class': 'huey.RedisHuey', # 使用RedisHery,PriorityRedisHuey支持任务优先级
'name': settings.DATABASES['default']['NAME'], # 使用 db 名称作为 huey name。
'results': True, # 存储任务的返回值。
'store_none': False, # 如果任务返回 None,则不要保存到结果中。
'immediate': settings.DEBUG, # 是否启用即时模式
'utc': True, # U在内部始终使用 UTC时间。
'connection': {
'host': 'localhost',
'port': 6379,
'db': 0
},
'consumer': {
'workers': 4,
'worker_type': 'thread',
'initial_delay': 0.1, # 最小轮询间隔 -d.
'max_delay': 10.0, # 最大轮询间隔, -m.
'scheduler_interval': 1, # 每秒检查一次调度, -s.
'periodic': True, # 启用 crontab 功能。
'check_worker_health': True, # 启用工作人员健康检查。
'health_check_interval': 1, # 每秒检查一次工作人员的健康状况。
},
}
# huey.py 在huey.py中导入我们的任务模块
from demo import tasks as demo_tasks
# __init__.py 初始化的时候自动加载我们的任务模块
from .huey import *
4.创建我们的定时任务
# tasks.py
from huey import crontab
from huey.contrib.djhuey import periodic_task, task
@task()
def count_beans(number):
print('-- counted %s beans --' % number)
return 'Counted %s beans' % number
@periodic_task(crontab(minute='*/5'))
def every_five_mins():
print('每五分钟由消费者打印一次')
5.运行消费者
./manage.py run_huey
三、有趣的任务类型
1.定期任务
Huey 提供了类似 crontab 的功能,使功能能够按给定的时间表自动执行。
在以下示例中,我们将声明periodic_task()每 3 分钟执行一次并在消费者进程 stdout 上打印一条消息:
from huey import crontab
from huey.contrib.djhuey import periodic_task, task
@task()
def add(a, b):
return a + b
@periodic_task(crontab(minute='*/3'))
def every_three_minutes():
print('This task runs every three minutes')
每分钟一次,调度程序将检查是否应该调用任何周期性任务。如果是这样,该任务将被排队执行。
crontab()函数接受以下参数:
- 分钟
- 小时
- 天
- 月
- day_of_week(0=星期日,6=星期六)
可接受的输入:
* - 始终为真,例如如果hour='*',则规则匹配任何时间。
*/n- 每n 个间隔,例如minute='*/15'表示每 15 分钟。
m-n- 每次都运行m..n包括在内。
m,n- 在m和n上运行。
多个规则可以通过用逗号分隔各个规则来表示,例如:
# 在 9点和 11点以及16点-18点之间每10分钟运行一次。
crontab(minute='*/10', hour='9-11,16-18')
2.重试失败的任务
有时我们可能有一个我们预期可能会不时失败的任务,在这种情况下我们应该重试它。Huey 支持自动重试任务给定的次数,可以选择在尝试之间有延迟。
在这里,我们将声明一个大约有一半时间失败的任务。要将此任务配置为自动重试,请使用装饰器的retries参数:task()
from huey.contrib.djhuey import task
import random
@task(retries=2) # Retry the task up to 2 times.
def flaky_task():
if random.randint(0, 1) == 0:
raise Exception('failing!')
当我们调用这个任务时会发生什么?
如果任务在第一次调用时失败,它将重试最多两次。
要指定重试尝试之间的延迟,我们可以添加一个retry_delay 参数。该任务将最多重试两次,两次尝试之间有 10 秒的延迟:
@task(retries=2, retry_delay=10)
def flaky_task():
# ...
3.任务优先级
任务优先级支持需要 Redis 5.0 或更高版本。要在 Redis 中使用任务优先级,请使用PriorityRedisHuey代替 RedisHuey。
Huey 任务可以被赋予优先级,让您确保您最重要的任务不会在工作人员忙碌时被延迟。
可以将优先级分配给任务函数,在这种情况下,任务的所有调用都将默认为给定的优先级。当没有给出优先级时,任务将默认为优先级0。
要了解它是如何工作的,让我们定义一个具有优先级 ( 10) 的任务:
@huey.task(priority=10)
def send_email(to, subj, body):
return mailer.send(to, 'webmaster@myapp.com', subj, body)
当我们调用这个任务时,它将在任何其他优先级小于 10 的待处理任务之前被处理。所以我们可以想象我们的队列看起来像这样:
- process_payment- 优先级 = 50
- check_spam- 优先级 = 1
- make_thumbnail- 优先级 = 0(默认)
调用send_email()任务:
send_email('new_user@foo.com', 'Welcome', 'blah blah')
现在待处理任务的队列将是:
- process_payment- 优先级 = 50
- send_email- 优先级 = 10
- check_spam- 优先级 = 1
- make_thumbnail- 优先级 = 0
任务优先级仅影响从待处理任务队列中拉出的任务的顺序。如果有一段时间您的员工无法跟上大量涌入的任务,Huey 的priority功能可以确保您最重要的任务不会被延迟。
还可以在计划将来运行的任务时指定特定于任务的优先级覆盖,我们可以指定优先级periodic_task:
@huey.periodic_task(crontab(minute='0', hour='*/3'), priority=10)
def some_periodic_task():
# ...
4.锁定任务
可以使用该Huey.lock_task()方法完成任务锁定,此锁可防止对任务的多次调用同时运行。
如果发生第二次调用并且无法获取锁,则 TaskLockedException引发特殊并且不会执行任务。如果任务配置为重试,则正常重试。
例子:
@huey.periodic_task(crontab(minute='*/5'))
@huey.lock_task('reports-lock') # Goes *after* the task decorator.
def generate_report():
# If a report takes longer than 5 minutes to generate, we do
# not want to kick off another until the prevIoUs invocation
# has finished.
run_report()
@huey.periodic_task(crontab(minute='0'))
def backup():
# Generate backup of code
do_code_backup()
# Generate database backup. Since this may take longer than an
# hour, we want to ensure that it is not run concurrently.
with huey.lock_task('db-backup'):
do_db_backup()
4.日志记录
Huey 使用标准库logging模块来记录有关任务执行和消费者活动的信息。消息被记录到huey命名空间,消费者特定的消息被记录到huey.consumer.
当消费者运行时,它会将默认值绑定StreamHandler()到 huey 命名空间,以便所有消息都记录到控制台。可以使用以下使用者选项配置使用者日志记录:
- -l FILE, --logfile=FILE- 记录到文件。
- -v, --verbose- 详细日志记录(包括调试级别)
- -q, --quiet- 最少的日志记录
- -S, --simple- 简单的记录格式(“时间消息”)
示例:
python .\manage.py run_huey --logfile=./logs/huey.log
5.开发过程调试
@huey.task()
def add(a, b):
return a + b
# Call the add() function in "un-decorated" form, skipping all
# the huey stuff:
add.call_local(3, 4) # Returns 7.
总结
以上是对huey这个任务队列工具的简单介绍。事实上huey还为我们提供了更多的功能,包括动态周期性任务,指数退避重试,以及取消或暂停定期任务等等。大家可以到Huey的gitee详细了解一下。
gitee Huey