django - Celery - 收到类型为 'core.tasks.scrape_dev_to' 的未注册任务

Question

尝试启动并运行基于 celery 的刮刀。 celery 工作程序似乎可以独立运行，但是当我也运行 celerybeat 服务器时，工作程序给了我这个关键错误。

  File "c:\users\myusername\.virtualenvs\django-news-scraper-dbqk-dk5\lib\site-packages\celery\worker\consumer\consumer.py", line 555, in on_task_received
    strategy = strategies[type_]
KeyError: 'core.tasks.scrape_dev_to'
[2020-10-04 16:51:41,231: ERROR/MainProcess] Received unregistered task of type 'core.tasks.scrape_dev_to'.
The message has been ignored and discarded.

我在 stackoverflow 上找到了很多类似的答案，但没有一个能解决我的问题。我会在最后列出我尝试过的事情。

项目结构:

核心 - 任务

newsscraper -celery.py -settings.py

任务:

import time
from newsscraper.celery import shared_task, task
from .scrapers import scrape

@task
def scrape_dev_to():
    URL = "https://dev.to/search?q=django"
    scrape(URL)
    return

设置.py:

INSTALLED_APPS = [
    'django.contrib.admin',
     ...
    'django_celery_beat',
    'core',
]
...
# I Added this setting while troubleshooting, got a new ModuleNotFound error for core.tasks
#CELERY_IMPORTS = (
#    'core.tasks',
#)
CELERY_BROKER_URL = 'redis://localhost:6379'
CELERY_BEAT_SCHEDULE = {
    "ScrapeStuff": {
        'task': 'core.tasks.scrape_dev_to',
        'schedule': 10  # crontab(minute="*/30")
    }
}

celery.py:

from __future__ import absolute_import, unicode_literals

import os

from celery import Celery

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'newsscraper.settings')

app = Celery('newsscraper')

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

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

当我为 celery 工作程序运行调试时，我发现 celery 没有注册我想要的任务 (scrape_dev_to)。 celery.py 中的 app.autodiscover_tasks() 调用不应该处理这个问题吗？这是输出:

  . celery.accumulate
  . celery.backend_cleanup
  . celery.chain
  . celery.chord
  . celery.chord_unlock
  . celery.chunks
  . celery.group
  . celery.map
  . celery.starmap

当我尝试将 core.tasks 添加到 CELERY_IMPORTS 设置时，我也会收到 ModuleNotFoundError 错误。这是我对问题所在的最佳猜测，但我不知道如何解决它。

我尝试过的事情:

1. 将 core.tasks 添加到 celery_imports 设置中。当我尝试运行 celerybeat 时，这会导致一个新错误:“没有名为‘core.tasks’的模块”。
2. 对任务中的名称进行硬编码:name='core.tasks.scrape_dev_to'
3. 调用worker时显式指定celery配置:celery -A newsscraper worker -l INFO -settings=celeryconfig
4. 使用我的导入(例如，从 newsscraper.celery 而不是从 celery)
5. 在包含任务的模块的 init.py 中添加一些配置代码(包含设置和 celery.py 的模块的 init.py 中已包含该代码)
6. Python manage.py 检查未发现任何问题
7. 显式调用 core.tasks 工作:celery -A core.tasks worker -l INFO

Answer 1

我遇到了同样的问题，这个设置为我解决了。

在您的设置中

CELERY_IMPORTS = [
    'app_name.tasks',
]

和

# app_name/tasks.py

from celery import shared_task

@shared_task
def my_task(*args, **kwargs):
    pass

imports 的文档引用.