收到类型 xx 的未注册任务。该消息已被忽略并丢弃

Question

我在 ubuntu EC2 节点上有一个 django 项目，它执行计算密集型长运行过程，通常需要 60 多秒。我需要缓存结果。我一直在阅读 http://www.caktusgroup.com/blog/2014/06/23/scheduling-tasks-celery/ and http://michal.karzynski.pl/blog/2014/05/18/setting-up-an-asynchronous-task-queue-for-django-using-celery-redis/ 和文档。我已经能够在命令行上完成一项基本任务，但现在我正在尝试将其作为 Django 脚本运行。

现在我的 django tp1 视图中的代码结构是：

from django.shortcuts import render
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from __future__ import absolute_import
from celery import shared_task

@csrf_exempt
def index(request):

    token = str(request.POST.get('token', False))
    calculator(token)
    return HttpResponse(token)

@shared_task
def calculator(token):

    # do calculation
    # store result in cache

    return

在命令行我运行:

(env1)ubuntu@ip-172-31-22-65:~/projects/tp$ celery --app=tp.celery:app worker --loglevel=INFO

在消息的最后我得到：

[2015-03-24 19:49:47,045: ERROR/MainProcess] Received unregistered task of type 'tp1.views.calculator'.
The message has been ignored and discarded.    Did you remember to import the module containing this task? Or maybe you are using relative imports?

我的tasks.py:

from __future__ import absolute_import    

from celery import shared_task    

@shared_task
def test(param):
    return 'The test task executed with argument "%s" ' % param

我怎样才能让它工作？

Answer 1

首先，如果您正在使用 autodiscover Celery 可能无法 autodiscover 您的 tasks.py 文件。您是否通过为 autodiscover 创建一个 celery_app 文件来配置它 according to the docs 您的任务：

# project/celery_app.py
from __future__ import absolute_import

import os

from celery import Celery

from django.conf import settings

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

app = Celery('project_name')

# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
# YOUR APPLICATION MUST BE LISTED IN settings.INSTALLED_APPS for this to work
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

此外，如果您依赖 autodiscover，它通常找不到位于未调用 tasks.py.

的模块中的内容

因此，除了检查您的配置外，我还会尝试将您 views.py 中的任务移至 tasks.py:

 # project/app_name/tasks.py
 from celery_app import app
 @app.task()
 def calculator(token):

    # do calculation
    # store result in cache

    return

最后，您可以从 tasks.py 导入此函数并在您的视图中调用它。

希望对您有所帮助。

收到类型 xx 的未注册任务。该消息已被忽略并丢弃

Received unregistered task of type xx. The message has been ignored and discarded

python

django

celery