python - 如何检测我是否在 Celery worker 中运行？

Question

有没有办法以编程方式确定正在导入/运行的当前模块是在 celery worker 的上下文中完成的？

我们已经决定在运行 Celery worker 之前设置一个环境变量，并在代码中检查这个环境变量，但我想知道是否有更好的方法？

Answer 1

从 celery 4.2 开始，您还可以通过在 worker_ready 上设置标志来执行此操作。信号

在 celery.py :

from celery.signals import worker_ready
app = Celery(...)
app.running = False

@worker_ready.connect
def set_running(*args, **kwargs):
    app.running = True

现在您可以使用全局应用程序实例检查您的任务
看看你是否在运行。这对于确定使用哪个记录器非常有用。