python - django celery 收到类型为 'print_test' 的未注册任务

Question

接续问题:celery periodic tasks not executing

我已经用 celery 设置了 django，例如:

drf_project/drf_project/celery.py

from __future__ import absolute_import, unicode_literals
from celery import Celery
import os

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'drf_project.settings')
app = Celery('drf_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

在drf_project/drf_project/init.py

from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app

__all__ = ['celery_app']

drf_project/user_management/tasks.py

from drf_project.celery import app
from time import strftime

@app.task
def print_test():
    print strftime('%Y-%m-%d %H:%M:%S')
    with open('abc.txt', 'ab+') as test_file:
        test_file.writeline(strftime('%Y-%m-%d %H:%M:%S'))

在drf_project/drf_project/settings.py

INSTALLED_APPS += ('django_celery_beat',)
CELERYBEAT_SCHEDULE = {
    "test_1": {
        "task": "tasks.print_test",
        "schedule": timedelta(seconds=2),
    },
}

我运行了两个 celery worker 并使用命令在不同的终端中进行了 beat:

celery -A drf_project worker -l info -E

和

celery -A drf_project beat -l info -S django

节拍每 2 秒向 worker 发送任务，例如:

[2016-11-28 12:25:19,314: INFO/MainProcess] Scheduler: Sending due task Importing contacts (print_test)

但是工作人员抛出如下错误:

[2016-11-28 12:24:57,551: ERROR/MainProcess] Received unregistered task of type 'print_test'. 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: u'[[], {}, {"chord": null, "callbacks": null, "errbacks": null, "chain": null}]' (77b) Traceback (most recent call last): File "/usr/local/lib/python2.7/dist-packages/celery/worker/consumer/consumer.py", line 549, in on_task_received strategy = strategies[type_] KeyError: 'print_test'

我可以在我的 worker 应用程序中看到这样的任务:

[tasks]
  . user_management.tasks.print_test

如何解决这个错误？

Answer 1

尝试在 CELERYBEAT_SCHEDULE 中设置任务的绝对路径，“user_management.tasks.print_test”

它应该是这样的:

INSTALLED_APPS += ('django_celery_beat',)
CELERYBEAT_SCHEDULE = {
    "test_1": {
        "task": "user_management.tasks.print_test",
        "schedule": timedelta(seconds=2),
    },
}