基于CentOS7+Flask+supervisor部署的huey+redis轻量级队列任务

如果仅仅看官方文档,必然会产生非常多困惑,看官方例程也会发现有很多问题,在爬完坑之后,笔者觉得有必要写一个简单教程为后人乘凉,本文凭记忆记录的,所以可能存在偏差,如果有任何问题,欢迎留言和指正!

最初笔者想使用celery做异步运算,但是奈何celery4在windows系统下不再被支持,只好放弃,在git上找了很多其他异步库,然后在里面挑选了更新最频繁且关注人数最多的huey库,但后续发现其实国内讨论这个库的少之又少(笑

异步队列的模型非常简单,一个生产者,一个消费者。在使用 @huey.task() 不用 schedule 函数调用时,生产者就是被 @huey.task() 修饰的函数,这样使用比较简单,只要调用一次函数,就会向队列产生推送一个任务,消费者是 huey/bin/huey_consumer.py 

安装Redis及huey

python安装库(windows和centos相同,需要注意是否需要改为pip3)

pip install huey
pip install redis

克隆git文件就可以找到huey文件夹,将整个文件夹复制到项目目录下即可

Windows下

1、官网下载redis可执行程序,安装,默认会加入系统服务,随开机自启,一劳永逸

2、可以下载rdm来使用gui查看redis信息

3、在python-console里面测试redis是否正常工作

>>> import redis
>>> r = redis.StrictRedis(host='localhost', port=6379, db=0)
>>> r.set('foo', 'bar')
True
>>> r.get('foo')
'bar'

4、git上下载huey的例程,按说明运行mini例程即可测试huey是否正常

CentOS下

1、在官网 http://download.redis.io/releases/ 找到你想要的版本,然后下载

wget http://download.redis.io/releases/redis-stable.tar.gz

2、解压安装一条龙

tar -zxvf redis-stable.tar.gz
yum install gcc
cd redis-stable
make
make install PREFIX=/usr/local/redis MALLOC=libc

每次执行一行,安装路径在

/usr/local/redis

3、修改配置文件 /usr/local/redis/bin/redis.conf ,使其可以后台运行,如果配置文件不存在,从下载的压缩包里面拷过来即可

将文件内daemonize no改为daemonize yes

4、在该目录下运行redis测试是否正常

./redis-server redis.conf

5、修改文件 /etc/systemd/system/redis.service 新增下文以设置开机启动

[Unit]
Description=redis-server
After=network.target
[Service]
Type=forking
ExecStart=/usr/local/redis/bin/redis-server /usr/local/redis/bin/redis.conf
PrivateTmp=true
[Install]
WantedBy=multi-user.target

按行执行以下命令

systemctl daemon-reload
systemctl start redis.service
systemctl enable redis.service

6、新增软链

ln -s /usr/local/redis/bin/redis-cli /usr/bin/redis

输入redis即可调用

代码范例

下文将展示在Flask中调用的写法,最简单的调用格式参考官方例程的mini即可,笔者使用了blueprint工厂函数的结构,蓝图的具体用法需要自行查阅

1、导入huey库  app/__init__.py 

from huey import RedisHuey
huey = RedisHuey()

2、修饰函数  app/my_func/task.py 

from app import huey

@huey.task()
def yourTask(): whatToDo

在你需要进行异步运算的py文件内从app导入huey函数(非huey库),然后在对应函数前增加修饰,括号内可以填入一些参数,如 retries=2, retry_delay=10 ,意为如果失败,最多重复尝试2次,期间间隔10s,关于其他参数和使用细节,可以参考官方文档

3、在 manage.py 中引入,以便后续调用

from app import huey

这个py文件就是有 app.run() 的文件,具体命名需要读者自己调整

4、开启消费者

python app/huey/bin/huey_consumer.py manage.huey -w 4 -c 30 -l log/huey.log

如果在服务器运行,可能需要改为python3来调用,如果顺利运行,则会显示“以下函数可被调用 yourTask() ”,-w 4意为四个工人,即四线程,-c 30意为30s的心跳包保活,其中作者非常自信,认为huey基本不可能会出现自杀的行为,-l path意为开启log,默认为info等级记录,如果是在windows运行,则需要开单独窗口持续运行,比如cmd,官方使用的是shell调用(可能需要引入python环境)

#!/bin/sh

python app/huey/bin/huey_consumer.py manage.huey -w 4 -c 30 -l log/huey.log

在当前目录下右键开启git bash,然后输入

./run_huey.sh

如果在服务器调用,不可能前台一直挂着shell,因此后文将会介绍如何后台运行,在检测是否正常时可以通过shell或者直接运行python来调试

5、开启生产者

本文使用的是Flask,即运行后在网页上产生运算任务,这时将会自动推送到huey队列,在消费者的窗口里面可以看到有任务加入和执行,以及执行总时间,如果在调试过程中不想使用huey,直接注释掉修饰即可,就会变成阻塞运行,其他不需要改变,非常简单

笔者所使用的文件结构如下(此结构的问题在终止部分有额外说明

|-project
  |-manage.py
  |-run_huey.sh
  |-log
    |-huey.log
  |-app
    |-__init__.py
    |-my_func
      |-__init__.py
      |-task.py
    |-huey
      |-xxxxxx

仅提及了本文涉及的文件,其中huey为整个库,不一一列举

基于supervisor的后台运行

在使用supervisor之前也尝试了nohup,发现既不够好用,也没有保活,于是安装supervisor

yum install -y supervisor
systemctl enable supervisord

然后修改 /etc/supervisord.conf 文件,将10-13行的第一个字符;去掉,然后端口修改为未被占用的端口,可以通过 netstat -lnp|grep xxxx 查看端口是否被占用,然后修改用户名密码!!!

[inet_http_server]
port=127.0.0.1:9001
username=用户名
password=密码

更新过配置需要重载配置文件,不过得先启动,否贼会报错 [Errno 2] No such file or directory: file: /usr/lib64/python2.7/socket.py

systemctl start supervisord
supervisorctl reload

在防火墙中开启对应端口(其中9001根据实际自行修改)

firewall-cmd --zone=public --add-port=9001/tcp --permanent
firewall-cmd --reload

在网址后加入:9001即可访问,如果显示服务器积极拒绝,则将127.0.0.1改为0.0.0.0,然后reload重载再尝试

配置后台任务

任务在 /etc/supervisord.d/ 下,以xxx.ini的文件存储

[program:进程名]
stdout_logfile=/log目录的绝对路径/huey_supervisor.out.log
stderr_logfile=/log目录的绝对路径/huey_supervisor.out.err
environment=PYTHONPATH="python项目的绝对路径"
directory=python项目的绝对路径
command=/usr/bin/python3 /消费者的绝对路径/huey_consumer.py manage.huey -w 1 -c 30 -l /log目录的绝对路径/huey.log
user=root
autostart=true
autorestart=true
redirect_stderr=true

这是配置的范例,其中红色部分需要根据实际情况自行填写,如果user=root注释掉,则需要将消费者的权限(可能还要其他的文件权限)放开,要不然会无法读取,所有路径都需要填写绝对路径,要不然无法读取,进程名和文件名无关

之后用命令 supervisorctl start xxxx 启动即可

supervisorctl命令集
> status           #查看程序状态
> stop name    #关闭name程序
> start name    #启动name程序
> restart name # 重启name程序
> reread          #读取有更新的配置文件,不会启动新添加的程序
> update          #重启配置文件修改过的程序

异步结果获取及状态查询

对python项目进行修改后需要重启huey进程以生效,建议使用git bash运行,ctrl+c终止后重新启动即可

官方文档对很多函数没有加以说明,建议直接看源码跳查,比较清晰直接

官方获取结果的方式有非常多,加上笔者使用的是redis,因此有更多方式获得结果,不过既然是使用异步运算,那么结果自然也是需要异步获取,标准添加任务方式返回的结果为自建类型<Result>,在异步情况下行不通,所以需要记录下 task_id ,尝试使用字符串切片,居然没问题

huey_id = str(result)[14:-1]

利用 result 和 id ,可以分别用官方提供的两个函数查询,其中 blocking 为阻塞查询(使用 huey 的函数查询后会删除掉 redis 中的记录)

result.get(blocking=True)
huey.result(huey_id, blocking=True)

利用 redis + id 可以使用 redis 库查询,其中结果是哈希值,所以使用 hget 来获取,在判空之后可以对结果使用 pickle 解码

conn = redis.Redis(host="127.0.0.1", port="6379")
conn.hget(name="huey.results.huey", key=huey_id)
pickle.loads(conn.hget(name="huey.results.huey", key=huey_id))

终止任务也有两个函数,分别为

result.revoke()
huey.revoke_by_id(huey_id, revoke_once=True)

终止之后git bash上会显示任务被终止,不再执行

但是官方函数库里面没有看到对任务状态和队列状态查询的函数,那就假设返回<Result>即加入队列成功,然后判断是否被终止来近似表示是否处于队列中,但没法判断是正在执行还是在排队,若读者找到对应的函数,欢迎留言

本应该也有两个函数,但是只在api里面找到一个,于是在 api.py 中仿写 revoke_by_id 完成另一个

result.is_revoked()
huey.is_revoked_by_id(huey_id)
def is_revoked_by_id(self, task_id, timestamp=None, peek=True):
return self.is_revoked(Task(id=task_id), timestamp, peek)

 到这里,上文代码的运行会不如你所愿,问题就在于上文文件结构中的 huey 文件夹,在项目下引入 huey 模块,有一定可能编辑器会直接从 site-package 里面引用 huey ,这将会导致在修改 api.py 后出现找不到函数的问题,所以,如果想要避免冲突,可以不在项目下加入 huey 文件夹,直接用绝对路径从 site-package 调用

此外,revoke函数仅仅支持在任务未运行的情况下被终止!!!如果函数已经开始运行,则无法被终止,可以采用干掉 worker 的方式来杀死进程,如果开了watchdog,将会在一定时间后 respawn ,此外,可以自行设立 flag ,在终止时修改flag以退出运算,但并不是很容易实现且效率低,取舍由读者自行把握,附上开发者的回复

 感谢以下GEEKS:

官方文档
https://huey.readthedocs.io/en/latest/index.html
Centos7安装Redis https:
//www.cnblogs.com/heqiuyong/p/10463334.html
centos7安装supervisor3.1.4
https://www.cnblogs.com/yjlch1016/p/10162918.html
原文地址:https://www.cnblogs.com/Pyrokine/p/12826718.html