python - celery 和信号

Question

我以前有过这样的功能

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

    if result > 4:
        result_higher_then_four.send(result)

    return result

result_higher_then_four 显然代表一个信号。

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

谢谢

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

    if result > 4:
        result_higher_then_four.send(result)

    return result

Answer 1

问题是信号接收器没有注册。 celery worker 在他们自己的进程中运行，因此需要在该进程中进行信号连接。如果您知道它们是什么或可以发现它们，则可以在任务初始化期间使用 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