Python使用 Beanstalkd 做异步任务处理的方法
作者:Young_Ji 发布时间:2021-01-06 18:18:19
使用 Beanstalkd 作为消息队列服务,然后结合 Python 的装饰器语法实现一个简单的异步任务处理工具.
最终效果
定义任务:
from xxxxx.job_queue import JobQueue
queue = JobQueue()
@queue.task('task_tube_one')
def task_one(arg1, arg2, arg3):
# do task
提交任务:
task_one.put(arg1="a", arg2="b", arg3="c")
然后就可以由后台的 work 线程去执行这些任务了。
实现过程
1、了解 Beanstalk Server
Beanstalk is a simple, fast work queue. https://github.com/kr/beanstalkd
Beanstalk 是一个 C 语言实现的消息队列服务。 它提供了通用的接口,最初设计的目的是通过异步运行耗时的任务来减少大量Web应用程序中的页面延迟。针对不同的语言,有不同的 Beanstalkd Client 实现。 Python 里就有 beanstalkc 等。我就是利用 beanstalkc 来作为与 beanstalkd server 通信的工具。
2、任务异步执行实现原理
beanstalkd 只能进行字符串的任务调度。为了让程序支持提交函数和参数,然后由woker执行函数并携带参数。需要一个中间层来将函数与传递的参数注册。
实现主要包括3个部分:
Subscriber: 负责将函数注册到 beanstalk 的一个tube上,实现很简单,注册函数名和函数本身的对应关系。(也就意味着同一个分组(tube)下不能有相同函数名存在)。数据存储在类变量里。
class Subscriber(object):
FUN_MAP = defaultdict(dict)
def __init__(self, func, tube):
logger.info('register func:{} to tube:{}.'.format(func.__name__, tube))
Subscriber.FUN_MAP[tube][func.__name__] = func
JobQueue: 方便将一个普通函数转换为具有 Putter 能力的装饰器
class JobQueue(object):
@classmethod
def task(cls, tube):
def wrapper(func):
Subscriber(func, tube)
return Putter(func, tube)
return wrapper
Putter: 将函数名、函数参数、指定的分组组合为一个对象,然后 json 序列化为字符串,最后通过 beanstalkc 推送到beanstalkd 队列。
class Putter(object):
def __init__(self, func, tube):
self.func = func
self.tube = tube
# 直接调用返回
def __call__(self, *args, **kwargs):
return self.func(*args, **kwargs)
# 推给离线队列
def put(self, **kwargs):
args = {
'func_name': self.func.__name__,
'tube': self.tube,
'kwargs': kwargs
}
logger.info('put job:{} to queue'.format(args))
beanstalk = beanstalkc.Connection(host=BEANSTALK_CONFIG['host'], port=BEANSTALK_CONFIG['port'])
try:
beanstalk.use(self.tube)
job_id = beanstalk.put(json.dumps(args))
return job_id
finally:
beanstalk.close()
Worker: 从 beanstalkd 队列中取出字符串,然后通过 json.loads 反序列化为对象,获得 函数名、参数和tube。最后从 Subscriber 中获得 函数名对应的函数代码,然后传递参数执行函数。
class Worker(object):
worker_id = 0
def __init__(self, tubes):
self.beanstalk = beanstalkc.Connection(host=BEANSTALK_CONFIG['host'], port=BEANSTALK_CONFIG['port'])
self.tubes = tubes
self.reserve_timeout = 20
self.timeout_limit = 1000
self.kick_period = 600
self.signal_shutdown = False
self.release_delay = 0
self.age = 0
self.signal_shutdown = False
signal.signal(signal.SIGTERM, lambda signum, frame: self.graceful_shutdown())
Worker.worker_id += 1
import_module_by_str('pear.web.controllers.controller_crawler')
def subscribe(self):
if isinstance(self.tubes, list):
for tube in self.tubes:
if tube not in Subscriber.FUN_MAP.keys():
logger.error('tube:{} not register!'.format(tube))
continue
self.beanstalk.watch(tube)
else:
if self.tubes not in Subscriber.FUN_MAP.keys():
logger.error('tube:{} not register!'.format(self.tubes))
return
self.beanstalk.watch(self.tubes)
def run(self):
self.subscribe()
while True:
if self.signal_shutdown:
break
if self.signal_shutdown:
logger.info("graceful shutdown")
break
job = self.beanstalk.reserve(timeout=self.reserve_timeout) # 阻塞获取任务,最长等待 timeout
if not job:
continue
try:
self.on_job(job)
self.delete_job(job)
except beanstalkc.CommandFailed as e:
logger.warning(e, exc_info=1)
except Exception as e:
logger.error(e)
kicks = job.stats()['kicks']
if kicks < 3:
self.bury_job(job)
else:
message = json.loads(job.body)
logger.error("Kicks reach max. Delete the job", extra={'body': message})
self.delete_job(job)
@classmethod
def on_job(cls, job):
start = time.time()
msg = json.loads(job.body)
logger.info(msg)
tube = msg.get('tube')
func_name = msg.get('func_name')
try:
func = Subscriber.FUN_MAP[tube][func_name]
kwargs = msg.get('kwargs')
func(**kwargs)
logger.info(u'{}-{}'.format(func, kwargs))
except Exception as e:
logger.error(e.message, exc_info=True)
cost = time.time() - start
logger.info('{} cost {}s'.format(func_name, cost))
@classmethod
def delete_job(cls, job):
try:
job.delete()
except beanstalkc.CommandFailed as e:
logger.warning(e, exc_info=1)
@classmethod
def bury_job(cls, job):
try:
job.bury()
except beanstalkc.CommandFailed as e:
logger.warning(e, exc_info=1)
def graceful_shutdown(self):
self.signal_shutdown = True
写上面代码的时候,发现一个问题:
通过 Subscriber 注册函数名和函数本身的对应关系,是在一个Python解释器,也就是在一个进程里运行的,而 Worker 又是异步在另外的进程运行,怎么样才能让 Worker 也能拿到和 Putter 一样的 Subscriber。最后发现通过 Python 的装饰器机制可以解决这个问题。
就是这句解决了 Subscriber 的问题
import_module_by_str('pear.web.controllers.controller_crawler')
# import_module_by_str 的实现
def import_module_by_str(module_name):
if isinstance(module_name, unicode):
module_name = str(module_name)
__import__(module_name)
执行 import_module_by_str 时, 会调用 __import__ 动态加载类和函数。将使用了 JobQueue 的函数所在模块加载到内存之后。当 运行 Woker 时,Python 解释器就会先执行 @修饰的装饰器代码,也就会把 Subscriber 中的对应关系加载到内存。
实际使用可以看 https://github.com/jiyangg/Pear/blob/master/pear/jobs/job_queue.py
来源:https://www.jianshu.com/p/cc9cd2892ff8
猜你喜欢
- 在网络浏览器软件中,可以Internet Explorer (IE)现在是一种标准的软件。可以看到,运行不同版本的Windows操作系统(和
- 使用方法:terminal中输入python weather.py http://www.weather.com.cn/weather/10
- 一般的防止被框架方式或者被框架后自动转向,方法如下: if (top.location != self.loc
- 给定一篇英语文章,要求统计出所有单词的个数,并按一定次序输出。思路是利用go语言的map类型,以每个单词作为关键字存储数量信息,代码实现如下
- <SCRIPT language=vbscript event=BeforeInitialBind(i
- 之前有看过一个博文写的是白社会的设计很好但运营却有些遭,因为对每一个WebGame的推出时间把握不准,会有几个应用同时上线造成影响力的冲突,
- 经常看见MOP上有人贴那种动态的图片,就是把一个字符串作为参数传给一个 * 页,就会生成一个带有这个字符串的图片,这个叫做文字水印。像什么原
- 本文实例讲述了PHP中PDO事务处理操作。分享给大家供大家参考,具体如下:概要:将多条sql操作(增删改)作为一个操作单元,要么都成功,要么
- 我们提倡无论何时都尽可能地使用CSS,这样我们更容易取得成功.现在浏览器对CSS的支持已经非常好,肯定足以让你用来控制你的网页布局与排版.但
- 如何让你的CSS代码更具有组织性和易维护性,为什么你的样式表总是臃肿和混乱的?有些时候是源于一开始书写时的混乱和草率,有时候也是因为后期的维
- 函数如下: function update_timelist(&$arr,$timestamp,$threshold){ $time
- 无参数函数先解释一下时间戳,所谓时间戳,即自1970年1月1日00:00:00所经历的秒数,然后就可以理解下面的函数了。下面代码默认from
- 内容摘要:下面是虚机维护中,经常碰到的一些ASP程序中的数据库调用的错误,现收集整理如下:1.不能打开注册表关键字(8007000e);2.
- CSS2.1 中规定了关于 CSS 规则 Specificity(特异性)的计算方式,用一个四位的数
- 本文实例为大家分享了Python KNN分类算法的具体代码,供大家参考,具体内容如下KNN分类算法应该算得上是机器学习中最简单的分类算法了,
- 代码如下:Class Vector Private vector_datas() Private&n
- 功能很简单,代码也很简洁,这里就不多废话了。package mainimport ( "fmt
- 同志们,经过不懈的努力,查了各种文档,终于鼓捣出了一个稍微像样一点的node项目,当然如果直接拿去项目里用,这个demo还太简单,毕竟一个完
- 简单低级的爬虫速度快,伪装度低,如果没有反爬机制,它们可以很快的抓取大量数据,甚至因为请求过多,造成服务器不能正常工作。而伪装度高的爬虫爬取
- 本文较为详细的讲述了Python中常用的模块,分享给大家便于大家查阅参考之用。具体如下:1.内置模块(不用import就可以直接使用)常用内