Python使用 Beanstalkd 做異步任務處理的方法
使用 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 線程去執(zhí)行這些任務了。
實現過程
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、任務異步執(zhí)行實現原理
beanstalkd 只能進行字符串的任務調度。為了讓程序支持提交函數和參數，然后由woker執(zhí)行函數并攜帶參數。需要一個中間層來將函數與傳遞的參數注冊。
實現主要包括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 中獲得 函數名對應的函數代碼，然后傳遞參數執(zhí)行函數。
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
寫上面代碼的時候，發(fā)現一個問題：
通過 Subscriber 注冊函數名和函數本身的對應關系，是在一個Python解釋器，也就是在一個進程里運行的，而 Worker 又是異步在另外的進程運行，怎么樣才能讓 Worker 也能拿到和 Putter 一樣的 Subscriber。最后發(fā)現通過 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)
執(zhí)行 import_module_by_str 時， 會調用 __import__ 動態(tài)加載類和函數。將使用了 JobQueue 的函數所在模塊加載到內存之后。當 運行 Woker 時，Python 解釋器就會先執(zhí)行 @修飾的裝飾器代碼，也就會把 Subscriber 中的對應關系加載到內存。
實際使用可以看 https://github.com/jiyangg/Pear/blob/master/pear/jobs/job_queue.py
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。
相關文章
Python 在處理數據方面非常出色,通常,數據集 會包括多個變量和許多實例,這使得很難理解數據的情況,數據可視化是幫助您識別數據模式的一種有用方式,本教程將描述如何使用 matplotlib 在 Python 中繪制數據,需要的朋友可以參考下2024-10-10
Python編寫可視化界面的全過程（Python+PyCharm+PyQt）
這篇文章主要給大家介紹了關于Python編寫可視化界面的相關資料,主要使用了Python+PyCharm+PyQt，文中通過實例代碼介紹的非常詳細，需要的朋友可以參考下2021-05-05
pip freeze 是 Python 包管理器 pip 的一個非常有用的命令,它可以列出當前 Python 環(huán)境中已經安裝的所有包及其版本,本文將詳細介紹 pip freeze 的用途及其在不同場景下的使用方法,需要的朋友可以參考下2024-08-08
這篇文章主要介紹了如何解決.cuda()加載用時很長的問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-05-05
這篇文章主要介紹了Python實現DHCP請求方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-06-06
這篇文章主要介紹了python對json的相關操作,結合實例形式詳細分析了json的概念、功能以及Python針對json的解析、輸出、排序、轉換等操作技巧,需要的朋友可以參考下2017-01-01
在本文中,我們將深入討論如何實現樹的序列化與反序列化算法,提供Python代碼實現,并詳細說明算法的原理和步驟,感興趣的可以了解一下2023-11-11
這篇文章主要介紹了python模仿手機通訊錄搜索功能，具有一定的參考價值，感興趣的小伙伴們可以參考一下2018-02-02
今天小編就為大家分享一篇Python讀寫/追加excel文件Demo，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-05-05
這篇文章主要介紹了Python實現的簡單發(fā)送郵件腳本分享,本文使用smtplib模塊實現郵件的發(fā)送,需要的朋友可以參考下2014-11-11