芹菜和信号

Question

我曾经有这样的功能

def calculate(self, input):
    result = input * 2

    if result > 4:
        result_higher_then_four.send(result)

    return result

result_higher_then_four 显然代表一个信号。

然后我介绍了 celery，我的功能如下所示，我再也没有收到信号。我想每个进程都绑定了信号，并且由于 celery 在不同的进程中运行，这意味着我无法在主进程中捕获信号。我应该使用thread_local 来解决这个问题吗？还是我忽略了显而易见的事情？

谢谢

@task
def calculate(self, input):
    result = input * 2

    if result > 4:
        result_higher_then_four.send(result)

    return result

Answer 1

您可以使用 celeryd_init 信号来初始化您的工作人员和信号 http://celery.readthedocs.org/en/latest/userguide/signals.html#celeryd-init

根据您提供的内容，我已经测试过：

from celery.signals import celeryd_init
from celery.utils.dispatch import Signal

def process_result(result, *args, **kwargs):
    print "signals received: %s" % result

result_higher_then_four = Signal()

@celeryd_init.connect
def init_signals(*args, **kwargs):
    result_higher_then_four.connect(process_result)

@task(bind=True)
def calculate(self, input):
    result = input * 2

    if result > 4:
       result_higher_then_four.send(result=result, sender=self)

    return result

Answer 2

问题是信号接收器没有注册。芹菜工人在他们自己的进程中运行，因此需要在该进程中建立信号连接。如果您知道它们是什么或可以发现它们，您可以在任务初始化期间使用this technique 注册它们。

当然，这首先消除了使用信号的一些好处，因为您需要提前了解连接。

一个想法是假设信号接收器将始终在每个应用程序的模型模块中注册。在这种情况下，以下将起作用。

class CalculateTask(celery.Task):

    def __init__(self):
        from django.conf import settings
        for app in settings.INSTALLED_APPS:
            app_models = '{}.{}'.format(app,'models') 
            __import__(app_models, globals=globals())                                 

    def run(self, input):
        result = input * 2
        if result > 4:
            result_higher_then_four.send(result)

        return result