Python使用 Beanstalkd 做异步任务处理的方法

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")

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

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

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()

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 <p></p><p><span>立即学习</span>“<a href="https://pan.quark.cn/s/00968c3c2c15" style="text-decoration: underline !important; color: blue; font-weight: bolder;" rel="nofollow" target="_blank">Python免费学习笔记（深入）</a>”；</p><p><strong>写上面代码的时候，发现一个问题：</strong></p><p>通过 Subscriber 注册函数名和函数本身的对应关系，是在一个Python解释器，也就是在一个进程里运行的，而 Worker 又是异步在另外的进程运行，怎么样才能让 Worker 也能拿到和 Putter 一样的 Subscriber。最后发现通过 Python 的装饰器机制可以解决这个问题。</p><p>就是这句解决了 Subscriber 的问题</p><p class="jb51code"></p><pre class="brush:py;">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)