Python Django Huey

Huey

---

---

前言

特性：

• 使用Python编写
• 仅支持Redis存储任务
• 支持Django，Flask

支持：

• 多进程执行任务
• 安排任务在给定时间或在给定延迟后执行
• 自动重试失败的任务
• 任务优先级
• 任务锁定

---

一、Huey是什么？

Huey是个Python任务队列，是个轻量级的替代品，没有多余的依赖关系。

二、使用步骤

1.安装

huey 可以使用pip从 PyPI 安装。

pip install huey

huey 在标准库之外没有依赖项，但是需要redis-py 才能利用 Redis 进行任务存储：

pip install redis

2.如何在我们的项目中进行配置

Huey带有与Django框架一起使用的特殊集成。该集成提供

• 通过Django settings模块配置Huey
• 将消费者作为Django的管理命令运行
• 自动发现 tasks.py 模块以简化任务的导入
• 正确管理数据库连接

支持的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, '[e-mail]', subj, body)

当我们调用这个任务时，它将在任何其他优先级小于 10 的待处理任务之前被处理。所以我们可以想象我们的队列看起来像这样：

• process_payment- 优先级 = 50
• check_spam- 优先级 = 1
• make_thumbnail- 优先级 = 0（默认）

调用send_email()任务：

send_email('[e-mail]', '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