使用 Beanstalkd 作為消息隊列服務，然后結合 Python 的裝飾器語法實現一個簡單的異步任務處理工具.
最終效果
定義任務:
from xxxxx.job_queue import JobQueuequeue = 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__] = funcJobQueue: 方便將一個普通函數轉換為具有 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
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持VEVB武林網。
新聞熱點
疑難解答
