python - celery 、Django 和@shared_task

Question

我正在尝试将 celery 与 Django 结合使用；我的一个应用程序中有一项任务，我想用 celery 运行它。不幸的是，我无法让 celery 找到任务，而是在运行 celery 的 shell 中收到以下错误消息:

bash% celery -A webflow worker -l info
-------------- celery@ws v4.4.2 (cliffs)
--- ***** ----- 
-- ******* ---- Linux-4.19.0-8-amd64-x86_64-with-debian-10.3 2020-05-10 11:49:09
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         webflow:0x7f3c8fb483c8
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 32 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . webflow.celery.debug_task

[2020-05-10 11:49:10,261: INFO/MainProcess] Connected to amqp://guest:**@127.0.0.1:5672//
[2020-05-10 11:49:10,267: INFO/MainProcess] mingle: searching for neighbors
[2020-05-10 10:47:28,942: ERROR/MainProcess] Received unregistered task of type 'simulation.tasks.unpack_simulation'.
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you're using relative imports?

Please see
http://docs.celeryq.org/en/latest/internals/protocol.html
for more information.

The full contents of the message body was:
'[["1b9944d2-f874-42d2-9dce-a0387c431b65"], {}, {"callbacks": null, "errbacks": null, "chain": null, "chord": null}]' (115b)
Traceback (most recent call last):
  File "/home/hove/sleipner/venv/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 562, in on_task_received
    strategy = strategies[type_]
KeyError: 'simulation.tasks.unpack_simulation'

django 项目名为 webflow它有一个名为 simulation 的 Django 应用程序- 总而言之，文件系统如下所示:

|── manage.py
├── requirements.txt
├── simulation                   // Django App
│   ├── admin.py
│   ├── apps.py
│   ├── __init__.py
│   ├── models.py
|   ├── api
|   |    │  
|   │    └── submit.py          // View which (tries) to invoke celery task
│   ├── tasks.py                // Module with celery task
└── webflow                     // Django Project
    ├── asgi.py
    ├── celery.py               // Celery app
    ├── __init__.py
    ├── settings.py
    ├── views.py                
    └── wsgi.py

文件的相关部分是:

模拟/tasks.py

from celery import shared_task
from simulation.models import Simulation

@shared_task
def unpack_simulation(sim_id):
    simulation = Simulation.objects.get(pk = sim_id)
    simulation.unpack()

webflow/celery.py

import os
from celery import Celery
from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'webflow.settings')
app = Celery('webflow')

app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks(settings.INSTALLED_APPS)

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

模拟/api/submit.py

import simulation.tasks


def submit(request):
    ...
    ...
    simulation.tasks.unpack_simulation.delay(sim_id)

webflow/__init__.py

from .celery import app as celery_app

__all__ = ('celery_app',)

这是我最好的逐字复制 https://docs.celeryproject.org/en/stable/django/first-steps-with-django.html ;我还尝试了许多在网上发现的微小变化 - 但没有成功。我试图掌握有关名称和导入的 celery 文档( https://docs.celeryproject.org/en/latest/userguide/tasks.html#task-naming-relative-imports ) - 这可能是解决方案；但我只能了解该文档 - 抱歉。

我认为我的设置没有完全损坏 - 如果我设置 CELERY_TASK_ALWAYS_EAGER = True在 webflow/settings.py归档它“有效”。

我使用的是 Django 3.0.?、Python 3.7 和 Celery 4.4.2。

Answer 1

答案:好的 - 这(正如预期的那样)100% 是我的错，但非常人为:在我运行 celery 的 shell 中，我有一个环境变量 DJANGO_SETTINGS_MODULE=some_other_project.settings