Django的芹菜收到類型 'print_test'

Question

與問題連續性的未註冊的任務：celery periodic tasks not executing

我已經建立了與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 -A drf_project worker -l info -E 

和

celery -A drf_project beat -l info -S django 

節拍正在發送任務的工人每2秒，如：

[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'

我可以看到任務在我的工人的應用程序，如：

[tasks] 
    . user_management.tasks.print_test 

如何解決這個錯誤？在CELERYBEAT_SCHEDULE

Answer 1

嘗試設置任務的絕對路徑，「user_management.tasks.print_test」

它應該是這樣的：

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