为什么 Celery 可以在 Python shell 中工作，但不能在我的 Django 视图中工作？ （导入问题）

Question

我安装了 Celery（最新的稳定版本。） 我有一个名为/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) 运行我的 Celery 工作人员，然后我试了一下。 我打开 Python 控制台并导入任务。然后，我运行 Submitter。

>>> 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."

这很奇怪，因为 celeryd 客户端在我启动它时确实显示它已注册。

[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

这就是我所做的，最终奏效了

在 Settings.py 我添加了

CELERY_IMPORTS = ("myapp.jobs", )

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

from celery.decorators import task

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

然后从命令行运行：python manage.py celeryd -l info

在另一个 shell 中我运行了 python manage.py shell，然后

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

我得到：

16

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

:-)

Answer 2

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

也可以尝试从 manage.py 启动 celeryd：

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

（-E 和 -B 可能需要也可能不需要，但这就是我使用的）。

Answer 3

请参阅文档中的自动命名和相对导入：

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", )