用户定义的 celery 任务类：在导入期间调用 init答案

【问题标题】：User defined celery task class : init is getting called during import用户定义的 celery 任务类：在导入期间调用 init
【发布时间】：2015-02-08 06:26:50
【问题描述】：

我正在尝试将 celery 任务用作一个类并查看以下行为。我想我错过了什么。让我先告诉你我想要达到的目标：
1. 用它的 init 函数创建一个类，它只会被 celery 调用一次。这将为我的课程设置所需的参数。我要在这里创建一个线程池。
2. 在生产者中创建这个 celery 任务对象的实例并将作业放入其中。

为了达到同样的效果，我尝试了 celery 网站上提到的简单示例并创建了一个示例类。我正在使用创建任务：

celery -c 1 -A proj worker --loglevel=debug

一开始它似乎在工作，但后来我观察到在 tester.py 中导入时会调用任务的 init，我可以通过传递标志在对象使用中停止这个 init，但导入期间的 init 在这里是一个真正的问题。

能否请您指出这个示例的正确用法。我不希望任务类的 init 被调用的次数超过我使用 celery 命令调用的次数。在现实生活中，它会创建不必要的线程。

如果可能的话，请指出一个最接近我上面提到的要求的例子。

芹菜.py

from __future__ import absolute_import
from celery import Celery
app = Celery('proj',
         broker='amqp://',
         backend='amqp://',
         include=['proj.tasks'])

# Optional configuration, see the application user guide.
app.conf.update(
    CELERY_TASK_RESULT_EXPIRES=3600,
)

if __name__ == '__main__':
    app.start()

tasks.py

from __future__ import absolute_import
from proj.celery import app

class NaiveAuthenticateServer(app.Task):

def __init__(self, celeryInst = 1):
    if celeryInst == 1:
        print "Hi, I am celery instant"
    else :
        print "Did you invoke me from command"
    self.users = {'george': 'password'}

def run(self, username, password):
    try:
        return self.users[username] == password
    except KeyError:
        return False

tester.py

from proj import tasks
obj = tasks.NaiveAuthenticateServer(0)
res = obj.delay('hi', 'hello')
print res.get()

tester.py 的o/p

你好，我是 celery Instant
你有没有从命令中调用我
假的

【问题讨论】：

标签： python multithreading celery distributed-computing

【解决方案1】：

您不应该自己创建任务类的实例，而是让 celery 在进程启动时自动为您完成。

因此，需要定义一个使用基类的任务函数：

@app.task(base=NaiveAuthenticateServer)
def my_task(arg1, arg2):
    print arg1, arg2

然后像这样提交任务：

from proj import tasks
tasks.my_task.delay('hi', 'hello')

【讨论】：