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

使用 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__] = 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 中获得 函数名对应的函数代码，然后传递参数执行函数。