python - celery 结果错误 "args must be a list or tuple"

Question

我正在运行一个 Django 网站并且刚刚让 Celery 运行，但是我遇到了令人困惑的错误。以下是代码的结构。

在测试.py中:

from tasks import *
from celery.result import AsyncResult

project = Project.objects.create()
# initalize various sub-objects of the project

c = function.delay(project.id)
r = AsyncResult(c.id).ready()
f = AsyncResult(c.id).failed()
# wait until the task is done  
while not r and not f:
    r = AsyncResult(c.id).ready()
    f = AsyncResult(c.id).failed()

self.assertEqual() #will fail because task fails

在 tasks.py 中:

from __future__ import absolute_import
from celery import shared_task

@shared_task
def function(project_id)
    #a bunch of calculations followed by a save of the project
    project = Project.objects.get(project=project_id)

    for part in project.part_set.all():
        partFunction(part.id)
        p = Part.objects.get(id=part.id)
        # add to various variables in project from variables in p
    project.save()

在 mainapp/settings.py 中:

BROKER_URL = "amqp://ipaddress"
CELERY_RESULT_BACKEND='amqp'
CELERY_ACCEPT_CONTENT = ['json','pickle','msgpack','yaml']
CELERY_IGNORE_RESULT = False

must by list/tuple 的 celery 调试控制台日志:

[INFO/MainProcess] Received task: myapp.tasks.function[id]
[ERROR/MainProcess] Task myapp.tasks.function[id]
    raised unexpected: ValueError('task args must be a list or tuple',)
Traceback:
   File "/python2.7/site-packages/celery/app/trace.py", line 240, in trace_task
       R = retval = fun(*args, **kwargs)
   File "/python2.7/site-packages/celery/app/trace.py", line 437, in __protected_call__
       return self.run(*args, **kwargs)
   File "/myapp/tasks.py", line 28, in function
       p = Part.objects.get(id=part.id)
   File "/python2.7/site-packages/celery/app/task.py", line 555, in apply_async
       **dict(self._get_exec_options(), **options)
   File "/python2.7/site-packages/celery/app/base.py", line 351, in send_task
       reply_to=reply_to or self.oid, **options
   File "celery/app/amqp.py", line 252, in publish_task
       raise ValueError('task args must be a list or tuple')
ValueError: task args must be a list or tuple

我得到的错误如上所示，AsyncResult(c.id).result: task args must be a list or tuple。这应该是一个简单的解决方案，但事实并非如此。当我把它列成这样时:

inline = [project.id]
c = function.delay(inline)

然后它改变主意并告诉我 AsyncResult(c.id).result: int() argument must be a string or a number, not 'list'

如您所想，我对可能出现的问题感到非常困惑。

---

编辑

任务.py

@shared_task
def function(app):
    @app.task(name='myapp.function', bind=True)
    def function(project_id):

测试.py

c = function.s(project.id).delay()

function.app 打印

Answer 1

您在任务中的代码中遇到错误，它显示在回溯中:

File "/myapp/tasks.py", line 28, in function
   p = Part.objects.get(id=part.id)

您的代码似乎是正确的，但从回溯来看，celery 似乎有一个旧版本的 task pickled。 非常重要 每当您更改 task.py 中的任何内容时，您都需要重新启动 celery(即使您更改了其他文件，但我不这么认为)。这可能是你问题的原因，它咬了我几次屁股。

也没有理由从数据库中单独提取 part p = Part.objects.get(id=part.id) 因为您已经获得了当前的零件实例当您在 for part in project.part_set.all(): 中迭代查询集时。这只是一个建议，您可能有更多代码需要执行该步骤。

作为旁注，如果它是您项目中的任务而不是某些可重用应用程序的一部分，只需使用常规 @task 装饰器，celery 会找到它，但请确保您配置Celery 应用程序正确，这是我的另一篇文章，可以指导您完成:Celery / Django Single Tasks being run multiple times

因此，只要您已正确配置所有内容，就可以像以前一样使用它:

@task #or @shared_task
def function(project_id)
    #a bunch of calculations followed by a save of the project
    project = Project.objects.get(project=project_id)
    ....

然后调用它:

result = function.delay(project.id)

或:

result = function.apply_async(args=(project.id,))

显然，我还建议通过在没有 celery function(project.id) 的情况下直接调用它来测试任务，但我相信你知道这一点。