2017-12-13

huey

Huey

简介

一个轻量级备选库

使用 python 编写
唯一依赖 Python Redis client

支持

多进程, 多线程或者 greenlet task 任务模型
可在指定时间或者延迟后执行的计划任务
像crontab一样定期重复执行
自动重试失败任务
存储任务结果

安装

pip

$ pip install huey

安装并自动启动数据库 SQLite/Redis支持

$ pip install huey[backends]

除了标准库依赖外, 只需要 redis

$ pip install redis

如果你的任务是 IO密集型 而不是 CPU密集型 参考IO/CPU bound, 那么可以考虑使用 greenlet 工作类型.

$ pip install gevent

使用 git 安装最新版本

$ git clone https://github.com/coleifer/huey.git
$ cd huey
$ python setup.py install

开始使用

向导

使用huey需要注意以下3点

生产者the produces(s) : i.e. web应用
消费者the consumer(s)
任务队列存储queue : e.g. Redis

intro

如图所示的这三个进程

左侧是生产者, 右上是消费者, 右下是队列

`建议`代码目录结构

.
├── __init__.py
├── app.py	# producer 生产者
├── config.py	# 队列存储配置文件
└── tasks.py	# 任务, 消费者

首先是配置你的任务队列. 消费者需要一个指明所用后端Huey实例

# config.py
from huey import RedisHuey
huey = RedisHuey()

huey对象封装了队列, 而这个队列用来负责存储和收取信息, 你的应用代码中的huey实例使用了后端来协调函数调用.

# tasks.py
from config import huey # import the huey we instantiated in config.py
# 通过装饰器, 可以简单的实现 消费者消费 功能
@huey.task()
def count_beans(num):
    print('-- counted %s beans --' % num)

上述例子说明了实现消费者的最简方法, 通过装饰器. 当count_beans被 main调用时, 只要enqueueing完成对函数的调用后会立即返回,而不是等待count_beans的执行结果,无论count_beans是否有耗时操作.

我们的主执行程序很简单, 只需要导入配置和任务即可. 这是为了确保当我们在通过指向配置运行consumer时, 任务也被导入并加载到了内存中.

# main.py
from config import huey  # import our "huey" object 需要保证导入名字为 huey, 否则会报错
from tasks import count_beans  # import our task
if __name__ == '__main__':
    beans = raw_input('How many beans? ')
    count_beans(int(beans))
    print('Enqueued job to count %s beans' % beans)

运行测试

注意:main.huey表示 main.py 中导入的 huey 实例
如果 from config import huey as mmhuey, 则运行脚本是 huey_consumer.py main.mmhuey

# 消费者, 消费队列 
huey_consumer.py main.huey
# 生产者, 压入队列
python main.py

任务结果

上述任务描述了 只发不收 的场景, 但是如果你需要任务的返回结果呢? 为了获取结果, 你只需要在你的任务中返回即可.

注意: 存储结果会浪费大量空间, 如果结果不是必须, 请返回为 None, 或者在 Huey实例中设置 result_store=False

# tasks.py
from config import huey
@huey.task()
def count_beans(num):
    print('-- counted %s beans --' % num)
    return 'Counted %s beans' % num

# main.py
from config import huey  # import our "huey" object 需要保证导入名字为 huey, 否则会报错
from tasks import count_beans  # import our task
if __name__ == '__main__':
    beans = raw_input('How many beans? ')
    rs = count_beans(int(beans))
    print('Enqueued job to count %s beans' % beans)
    print(rs()) # 由于使用异步队列, 故与使用显示调用的 rs.get() 一样都是 None
    # 如果需要获取结果 需要使用 rs(blocking=True)

在将来某个时刻运行

需要在 main 中调用时, 启用 schedule 来实现

# main.py
# 建立1个10s后执行的任务
count_beans.schedule(args=(100,), delay=10)
# 建立一个 60s 后运行的任务
count_beans.schedule(args=(100,),
                     eta=datetime.datetime.now() + datetime.timedelta(seconds=60))

失败重试机制

直接在 tasks 中使用装饰器

# tasks.py
@huey.task(retries=3)
def try_thrice():
    print('trying....')
    raise Exception('nope')

计划执行任务

使用 crontab 来实现.

# tasks.py
from datetime import datetime
from huey import crontab
from config import huey
@huey.periodic_task(crontab(minute='*'))
def print_time():
    print(datetime.now())

取消/暂停任务

普通任务

取消单个实例

res = count_beans(1024)
# 取消,尚未执行的任务
res.revoke()
# 当然也可以重新恢复
res.restore()

取消给定任务的所有实例

# 撤消
count_beans.revoke()
# 恢复
count_beans.restore()

定期任务

@huey.periodic_task(crontab(minute='*'))
def print_time():
    print(datetime.now())

停止一次

print_time.revoke(revoke_once=True)

停止到直到某个时间

# prevent printing time for 10 minutes
now = datetime.datetime.utcnow()
in_10 = now + datetime.timedelta(seconds=600)
print_time.revoke(revoke_until=in_10)

永远停止

print_time.revoke()

随时恢复

print_time.restore()

本文标题:huey

文章作者:coghost

发布时间:2017-12-13, 00:00:00

最后更新:2017-12-14, 10:18:33

原始链接:http://doequitable.com/20171213/python/tlibs/huey/

许可协议: "署名-非商用-相同方式共享 4.0" 转载请保留原文链接及作者。