芹菜和信號

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