时间戳

首页 > 解决方案 > 芹菜：收到“celery_deploy.tasks._stocks”类型的未注册任务

python - 芹菜：收到“celery_deploy.tasks._stocks”类型的未注册任务

问题描述

当我如下运行芹菜时： 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'

我在项目书签目录中的 celery.py（Django 项目 settings.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 应用程序 celery_deploy 中的 tasks.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)

我阅读了许多与我类似的问题，但仍然无法弄清楚。

我该怎么做才能解决它？谢谢。

标签： pythondjangocelery

解决方案

该错误可能与您的Instantiaion有关。假设您正在尝试App-Wide usage，那么以下内容将不起作用，因为您需要设置task_cls属性。

这里

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

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

您可以更改该行以包含您的文件名，例如

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

这应该有望解决问题。

推荐阅读