芹菜：收到类型为"celery_deploy.任务._股票"的未注册任务

Question

当我运行芹菜时，如下所示：celery -A bookmarks worker -l INFO

它返回：

[2020-11-10 23:14:21,649: ERROR/MainProcess] Received unregistered task of type 'celery_deploy.tasks._stocks'.
The message has been ignored and discarded.
Did you remember to import the module containing this task?
Or maybe you're using relative imports?
KeyError: 'celery_deploy.tasks._stocks'

项目书签目录中的mycelery.py(Django项目设置.py所在的位置(：

from __future__ import  absolute_import, unicode_literals
import os
from celery import Celery, platforms
from datetime import timedelta
from . import settings
BROKER_URL = 'redis://localhost:6379/0'
BACKEND_URL = 'redis://localhost:6379/1'
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'bookmarks.settings')
app = Celery('bookmarks', backend=BACKEND_URL, broker=BROKER_URL)
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks(lambda : settings.INSTALLED_APPS)
platforms.C_FORCE_ROOT = True
app.conf.update(
CELERY_ACKS_LATE = True,
CELERY_ACCEPT_CONTENT = ['json'],
CELERY_FORCE_EXECV = True,
CELERY_CONCURRENCY = 4,
CELERYD_MAX_TASKS_PER_CHILD = 10,
CELERYBEAT_SCHEDULE = {
'get_stock_': {
'task': 'celery_deploy.tasks._stocks',
'schedule': timedelta(seconds=21),
},
},
)
@app.task(bind = True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))

Django应用程序cele_deploy:中的我的任务.py

from __future__ import absolute_import
from celery import shared_task
from django.core.cache import cache
from urllib.request import urlopen
from bookmarks.celery import app
import json
import redis
apibase = "https://financialmodelingprep.com/api/v3/"
apikey = "[KEY]"
exchange_open_day = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri']
json_parsed = {}
CACHE_TIMEOUT = 60*60
@app.task
def _stocks():
url = apibase + "quotes/nyse" + "?" + apikey
response = urlopen(url)
stocks_parsed = json.loads(response.read().decode("utf-8"))
for _object in stocks_parsed:
cache.set(_object['symbol'], _object['price'], CACHE_TIMEOUT)

我读了许多与我相似的问题，但仍然搞不清楚。

我该怎么修？非常感谢。

Answer 1

错误可能与您的Instantiaion有关。假设您正在尝试应用程序范围的使用，那么以下操作将不起作用，因为您需要设置task_cls属性。

此处

app = Celery('bookmarks', backend=BACKEND_URL, broker=BROKER_URL)

bookmarks引用了您的目录，但在运行时无法导入所需的方法。

你能更改行以包括你的文件名吗，例如

app = Celery('bookmarks.tasks', backend=BACKEND_URL, broker=BROKER_URL)

这应该有望解决问题。