芹菜和信号

Question

我曾经有过这样的

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

    if result > 4: 
     result_higher_then_four.send(result) 

    return result 

其中result_higher_then_four显然代表了信号的功能。

然后我介绍芹菜和我的功能看起来像下面，我再也没有收到信号。我想信号是按每个进程绑定的，因为芹菜在不同的进程中运行，这意味着我无法在主进程中捕捉到信号。我应该使用thread_local来解决这个问题吗？还是我忽略了显而易见的？

感谢

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

    if result > 4: 
     result_higher_then_four.send(result) 

    return result 

Answer 1

问题是信号接收器没有被注册。芹菜工人在自己的过程中运行，因此需要在这个过程中建立信号连接。如果您知道它们是什么或可以发现它们，则可以在任务初始化期间使用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 

Answer 2

如果我理解正确，您想同样的进程发送ANS接收它发出的信号？如果是这样，为什么不使用：

os.kill(os.getpid(), signal.SIGUSER1) 

并相应地定义SIGUSR1的处理程序？

如果你想要另一个进程得到它，你必须有它的pid来发送信号，所以只需使用我在这里给出的相同命令而不是os.getpid（）。 除非我错过了什么？

Answer 3

可以使用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