Python で Beanstalkd を使用して非同期タスク処理を行う方法

この記事では、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")

これで、バックグラウンド作業スレッドがこれらのタスクを実行できるようになります。

実装プロセス

1. Beanstalkサーバーを理解する

Beanstalkは、C言語で実装されたシンプルで高速なワークキューサービスです。 これは共通のインターフェイスを提供し、元々は時間のかかるタスクを非同期で実行することで大規模な Web アプリケーションのページ遅延を削減するために設計されました。言語ごとに異なる Beanstalkd クライアント実装があります。 Pythonにはbeanstalkcなどがあります。私は Beantalkd サーバーと通信するためのツールとして Beantalkc を使用します。

2. 非同期タスク実行の実装原理

beanstalkdは文字列タスクのスケジューリングのみを実行できます。プログラムが関数とパラメーターの送信をサポートするために、関数はワーカーによって実行され、パラメーターが伝えられます。渡されたパラメータを使用して関数を登録するには、中間層が必要です。

実装には主に 3 つの部分が含まれます:

サブスクライバー: Beanstalk のチューブに関数を登録する責任を負います。実装は非常に単純で、関数名と関数自体の間の対応関係を登録します。 (同じグループ(チューブ)内に同じ関数名が存在できないことを意味します)。データはクラス変数に格納されます。

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でシリアル化します文字列に変換され、最終的に Beantalkc を通じて Beantalkd キューにプッシュされます。

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

ワーカー: Beantalkd キューから文字列を取得し、json.loads を通じてオブジェクトに逆シリアル化して、関数名、パラメーター、チューブを取得します。最後に、関数名に対応する関数コードをサブスクライバから取得し、パラメータを渡して関数を実行します。

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={&#39;body&#39;: 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(&#39;tube&#39;)
  func_name = msg.get(&#39;func_name&#39;)
  try:
   func = Subscriber.FUN_MAP[tube][func_name]
   kwargs = msg.get(&#39;kwargs&#39;)
   func(**kwargs)
   logger.info(u&#39;{}-{}&#39;.format(func, kwargs))
  except Exception as e:
   logger.error(e.message, exc_info=True)
  cost = time.time() - start
  logger.info(&#39;{} cost {}s&#39;.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 インタープリターで実行されます。ワーカーは別のプロセスで非同期に実行されますが、ワーカーもパターと同じサブスクライバーを取得するにはどうすればよいでしょうか?最後に、この問題は 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を使用した関数を含むモジュールをメモリにロードした後。 Wker を実行すると、Python インタープリターは最初に @-decorated デコレーター コードを実行し、Subscriber の対応する関係をメモリに読み込みます。

実際の使用方法については、https://github.com/jiyangg/Pear/blob/master/pear/jobs/job_queue.pyを参照してください

関連推奨事項:

php-beanstalkdメッセージキュークラスインスタンスの詳細な説明

以上がPython で Beanstalkd を使用して非同期タスク処理を行う方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。