芹菜相当于 JoinableQueue答案

【问题标题】：Celery equivalent of a JoinableQueue芹菜相当于 JoinableQueue
【发布时间】：2017-01-17 00:03:35
【问题描述】：

Celery 相当于 multiprocessing.JoinableQueue（或 gevent.queue.JoinableQueue）？

我正在寻找的功能是能够.join()来自发布者的 Celery 任务队列，等待队列中的所有任务完成。

等待初始 AsyncResult 或 GroupResult 是不够的，因为队列会由工作人员自己动态填满。

【问题讨论】：

您能否澄清一下，为什么group(*tasks) > apply > join 不够？工人如何影响您的小组结果？为什么你不能处理.join 结果？
@Slam 因为那只会等待完成初始任务组。工作人员不会影响组结果，他们会将更多任务添加到队列中。我的意图是等待队列中的所有任务完成（类似于我可以join() 和JoinableQueue 的方式，它会等待队列中的每个任务调用task_done()）。我可能可以通过在 Redis 中创建一个额外的共享计数器和一个 pub/sub 来实现这一点，但我想知道是否有一个更清洁、更可靠的方法只使用 Celery。

标签： python queue celery python-multiprocessing gevent

【解决方案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 而不是 pub/sub，因为我只需要一个监听器来监听“所有任务完成”事件（发布者）。

使用 JoinableCeleryQueue 非常简单 - begin() 在将任何任务添加到队列之前，使用常规 Celery API 添加任务，.join() 以等待所有任务完成。

【讨论】：