Celery 相当于一个 JoinableQueue
Celery equivalent of a JoinableQueue
Celery 的 multiprocessing.JoinableQueue
(or gevent.queue.JoinableQueue
) 等价物是什么?
我正在寻找的功能是 .join()
来自发布者的 Celery 任务队列,等待队列中的所有任务完成。
等待初始 AsyncResult
或 GroupResult
是不够的,因为队列由工作人员自己动态填满。
它可能并不完美,但这是我最终想到的。
它基本上是一个基于共享 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
而不是 pub/sub,因为我只需要一个监听器来监听 "all task finished" 事件(发布者)。
使用 JoinableCeleryQueue
非常简单 - begin()
在将任何任务添加到队列之前,使用常规 Celery 添加任务 API,.join()
等待所有待完成的任务。
Celery 的 multiprocessing.JoinableQueue
(or gevent.queue.JoinableQueue
) 等价物是什么?
我正在寻找的功能是 .join()
来自发布者的 Celery 任务队列,等待队列中的所有任务完成。
等待初始 AsyncResult
或 GroupResult
是不够的,因为队列由工作人员自己动态填满。
它可能并不完美,但这是我最终想到的。
它基本上是一个基于共享 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
而不是 pub/sub,因为我只需要一个监听器来监听 "all task finished" 事件(发布者)。
使用 JoinableCeleryQueue
非常简单 - begin()
在将任何任务添加到队列之前,使用常规 Celery 添加任务 API,.join()
等待所有待完成的任务。