快捷導(dǎo)航

Python使用 Beanstalkd 做異步任務(wù)處理的方法

更新時(shí)間：2018年04月24日 11:49:33 作者：Young_Ji

這篇文章主要介紹了Python使用 Beanstalkd 做異步任務(wù)處理的方法，小編覺得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過來看看吧

使用 Beanstalkd 作為消息隊(duì)列服務(wù)，然后結(jié)合 Python 的裝飾器語法實(shí)現(xiàn)一個(gè)簡(jiǎn)單的異步任務(wù)處理工具.

最終效果

定義任務(wù):

from xxxxx.job_queue import JobQueue

queue = JobQueue()

@queue.task('task_tube_one')
def task_one(arg1, arg2, arg3):
 # do task

提交任務(wù):

task_one.put(arg1="a", arg2="b", arg3="c")

然后就可以由后臺(tái)的 work 線程去執(zhí)行這些任務(wù)了。

實(shí)現(xiàn)過程

1、了解 Beanstalk Server

Beanstalk is a simple, fast work queue. https://github.com/kr/beanstalkd

Beanstalk 是一個(gè) C 語言實(shí)現(xiàn)的消息隊(duì)列服務(wù)。它提供了通用的接口，最初設(shè)計(jì)的目的是通過異步運(yùn)行耗時(shí)的任務(wù)來減少大量Web應(yīng)用程序中的頁面延遲。針對(duì)不同的語言，有不同的 Beanstalkd Client 實(shí)現(xiàn)。 Python 里就有 beanstalkc 等。我就是利用 beanstalkc 來作為與 beanstalkd server 通信的工具。

2、任務(wù)異步執(zhí)行實(shí)現(xiàn)原理

beanstalkd 只能進(jìn)行字符串的任務(wù)調(diào)度。為了讓程序支持提交函數(shù)和參數(shù)，然后由woker執(zhí)行函數(shù)并攜帶參數(shù)。需要一個(gè)中間層來將函數(shù)與傳遞的參數(shù)注冊(cè)。

實(shí)現(xiàn)主要包括3個(gè)部分:

Subscriber: 負(fù)責(zé)將函數(shù)注冊(cè)到 beanstalk 的一個(gè)tube上,實(shí)現(xiàn)很簡(jiǎn)單，注冊(cè)函數(shù)名和函數(shù)本身的對(duì)應(yīng)關(guān)系。(也就意味著同一個(gè)分組(tube)下不能有相同函數(shù)名存在)。數(shù)據(jù)存儲(chǔ)在類變量里。

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: 方便將一個(gè)普通函數(shù)轉(zhuǎn)換為具有 Putter 能力的裝飾器

class JobQueue(object):
 @classmethod
 def task(cls, tube):
  def wrapper(func):
   Subscriber(func, tube)
   return Putter(func, tube)

  return wrapper

Putter: 將函數(shù)名、函數(shù)參數(shù)、指定的分組組合為一個(gè)對(duì)象，然后 json 序列化為字符串，最后通過 beanstalkc 推送到beanstalkd 隊(duì)列。

class Putter(object):
 def __init__(self, func, tube):
  self.func = func
  self.tube = tube

 # 直接調(diào)用返回
 def __call__(self, *args, **kwargs):
  return self.func(*args, **kwargs)

 # 推給離線隊(duì)列
 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 隊(duì)列中取出字符串，然后通過 json.loads 反序列化為對(duì)象，獲得函數(shù)名、參數(shù)和tube。最后從 Subscriber 中獲得函數(shù)名對(duì)應(yīng)的函數(shù)代碼，然后傳遞參數(shù)執(zhí)行函數(shù)。

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) # 阻塞獲取任務(wù)，最長等待 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

寫上面代碼的時(shí)候，發(fā)現(xiàn)一個(gè)問題：

通過 Subscriber 注冊(cè)函數(shù)名和函數(shù)本身的對(duì)應(yīng)關(guān)系，是在一個(gè)Python解釋器，也就是在一個(gè)進(jìn)程里運(yùn)行的，而 Worker 又是異步在另外的進(jìn)程運(yùn)行，怎么樣才能讓 Worker 也能拿到和 Putter 一樣的 Subscriber。最后發(fā)現(xiàn)通過 Python 的裝飾器機(jī)制可以解決這個(gè)問題。

就是這句解決了 Subscriber 的問題

import_module_by_str('pear.web.controllers.controller_crawler')

# import_module_by_str 的實(shí)現(xiàn)
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 時(shí)，會(huì)調(diào)用 __import__ 動(dòng)態(tài)加載類和函數(shù)。將使用了 JobQueue 的函數(shù)所在模塊加載到內(nèi)存之后。當(dāng) 運(yùn)行 Woker 時(shí)，Python 解釋器就會(huì)先執(zhí)行 @修飾的裝飾器代碼，也就會(huì)把 Subscriber 中的對(duì)應(yīng)關(guān)系加載到內(nèi)存。

實(shí)際使用可以看 https://github.com/jiyangg/Pear/blob/master/pear/jobs/job_queue.py

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: