为什么Celery在Python shell中工作，但不在我的Django视图中工作？ （进口问题）

Question

我安装了芹菜（最新稳定版）。 我有一个目录叫/home/myuser/fable/jobs。在这个目录里，我有一个名为tasks.py文件：

from celery.decorators import task 
from celery.task import Task 

class Submitter(Task): 
    def run(self, post, **kwargs): 
     return "Yes, it works!!!!!!" 

在这个目录里，我也有一个名为celeryconfig.py：

BROKER_HOST = "localhost" 
BROKER_PORT = 5672 
BROKER_USER = "abc" 
BROKER_PASSWORD = "xyz" 
BROKER_VHOST = "fablemq" 

CELERY_RESULT_BACKEND = "amqp" 
CELERY_IMPORTS = ("tasks",) 

在我/etc/profile，我有这些设置我的PYTHONPATH：

• PYTHONPATH=/home/myuser/fable:/home/myuser/fable/jobs

所以我使用控制台（$ celeryd --loglevel=INFO）运行我的芹菜工作人员，然后我尝试一下。 我打开Python控制台并导入任务。然后，我运行提交者。

>>> import fable.jobs.tasks as tasks 
>>> s = tasks.Submitter() 
>>> s.delay("abc") 
<AsyncResult: d70d9732-fb07-4cca-82be-d7912124a987> 

一切正常，你可以在我的控制台中看到

[2011-01-09 17:30:05,766: INFO/MainProcess] Task tasks.Submitter[d70d9732-fb07-4cca-82be-d7912124a987] succeeded in 0.0398268699646s: 

但是当我进入我的Django的views.py并运行确切3行代码如上，我得到这个：

[2011-01-09 17:25:20,298: ERROR/MainProcess] Unknown task ignored: "Task of kind 'fable.jobs.tasks.Submitter' is not registered, please make sure it's imported.": {'retries': 0, 'task': 'fable.jobs.tasks.Submitter', 'args': ('abc',), 'expires': None, 'eta': None, 'kwargs': {}, 'id': 'eb5c65b4-f352-45c6-96f1-05d3a5329d53'} 
Traceback (most recent call last): 
    File "/home/myuser/mysite-env/lib/python2.6/site-packages/celery/worker/listener.py", line 321, in receive_message 
    eventer=self.event_dispatcher) 
    File "/home/myuser/mysite-env/lib/python2.6/site-packages/celery/worker/job.py", line 299, in from_message 
    eta=eta, expires=expires) 
    File "/home/myuser/mysite-env/lib/python2.6/site-packages/celery/worker/job.py", line 243, in __init__ 
    self.task = tasks[self.task_name] 
    File "/home/myuser/mysite-env/lib/python2.6/site-packages/celery/registry.py", line 63, in __getitem__ 
    raise self.NotRegistered(str(exc)) 
NotRegistered: "Task of kind 'fable.jobs.tasks.Submitter' is not registered, please make sure it's imported." 

这很奇怪，因为芹菜客户端在启动时显示它已注册。

[2011-01-09 17:38:27,446: WARNING/MainProcess] 
Configuration -> 
    . broker -> amqp://GOGOme@localhost:5672/fablemq 
    . queues -> 
     . celery -> exchange:celery (direct) binding:celery 
    . concurrency -> 1 
    . loader -> celery.loaders.default.Loader 
    . logfile -> [stderr]@INFO 
    . events -> OFF 
    . beat -> OFF 
    . tasks -> 
     . tasks.Decayer 
     . tasks.Submitter 

有人可以帮忙吗？

Answer 1

我相信你的tasks.py文件需要在django应用程序（在settings.py中注册）才能导入。或者，您可以尝试从主项目或其中一个应用程序导入__init__.py文件中的任务。

也可以尝试从manage.py开始celeryd：

$ python manage.py celeryd -E -B -lDEBUG 

（-E和-B可能会或可能不会是必要的，但这是我使用）。

Answer 2

请参阅自动命名和相对导入，在文档：

http://celeryq.org/docs/userguide/tasks.html#automatic-naming-and-relative-imports

任务名称为“tasks.Submitter”（如celeryd输出中列出）， 但导入任务，“ fable.jobs.tasks.Submitter“

我想这里最好的解决方案是，如果工作人员也把它看作”fable.jobs.tasks.Submitter“， 它从应用程序的角度来看更有意义。

CELERY_IMPORTS = ("fable.jobs.tasks",) 

Answer 3

这是我做这最后的设置工作

。PY我加

CELERY_IMPORTS = ("myapp.jobs",) 

MyApp文件夹下，我创建了一个名为jobs.py

from celery.decorators import task 

@task(name="jobs.add") 
def add(x, y): 
    return x * y 

然后从命令行运行文件：蟒蛇manage.py在另一个shell我跑celeryd -l信息

蟒蛇manage.py壳，然后

>>> from myapp.jobs import add 
>>> result = add.delay(4, 4) 
>>> result.result 

和我得到：

重要的一点是，当你添加一个新函数时你必须重新运行这两个命令shell。您必须在客户端和服务器上注册名称。

:-)