JoinableQueueのセロリ等価

Question

セロリはmultiprocessing.JoinableQueue（またはgevent.queue.JoinableQueue）に相当しますか？

私が探している機能は、キュー内のすべてのタスクが完了するのを待って、パブリッシャからセロリタスクキュー.join()にアクセスできることです。

待ち行列がワーカー自身によって動的に満杯になるので、最初の待ちであるAsyncResultまたはGroupResultは、十分ではありません。

Answer 1

完璧ではないかもしれませんが、これは私が最終的に思いついたものです。

共有Redisカウンタとリストリスナーに基づいて、基本的には既存のCeleryキューの上にJoinableQueueラッパーです。キュー名はルーティングキーと同じにする必要があります（before_task_publishとtask_postrun信号の内部実装の詳細が原因です）。

joinableceleryqueue.py：

from celery.signals import before_task_publish, task_postrun 
from redis import Redis 
import settings 

memdb = Redis.from_url(settings.REDIS_URL) 

class JoinableCeleryQueue(object): 
    def __init__(self, queue): 
     self.queue = queue 
     self.register_queue_hooks() 

    def begin(self): 
     memdb.set(self.count_prop, 0) 

    @property 
    def count_prop(self): 
     return "jqueue:%s:count" % self.queue 

    @property 
    def finished_prop(self): 
     return "jqueue:%s:finished" % self.queue 

    def task_add(self, routing_key, **kw): 
     if routing_key != self.queue: 
      return 

     memdb.incr(self.count_prop) 

    def task_done(self, task, **kw): 
     if task.queue != self.queue: 
      return 

     memdb.decr(self.count_prop) 
     if memdb.get(self.count_prop) == "0": 
      memdb.rpush(self.finished_prop, 1) 

    def register_queue_hooks(self): 
     before_task_publish.connect(self.task_add) 
     task_postrun.connect(self.task_done) 

    def join(self): 
     memdb.brpop(self.finished_prop) 

私は私が唯一のイベント（出版社）、「すべてのタスクが終了した」を聴いてのリスナーを必要とするよう代わりにパブリッシュ/サブスクライブのBRPOPを使用することを選択しました。

JoinableCeleryQueueを使用することは非常に簡単です - begin()キューにタスクを追加する前に、通常のCelery API .join()を使用してタスクを追加して、すべてのタスクが完了するまで待ちます。