django celery 和弦任务错误

Question

我在尝试使用chord发送任务时遇到了错误，这个错误发生在我发送完所有任务之后。

Traceback (most recent call last):
  File "<console>", line 1, in <module>
  File "/Users/wenzhixue/projects/workspace/fallfor_core/twitter/tasks.py", line 13, in bulk_change_bio
    chord([change_bio_task.delay(account,'http://fallfor.com') for account in account_list ])(shutdown.s(c))
  File "/Library/Python/2.7/site-packages/celery/canvas.py", line 470, in __call__
    _chord = self.type
  File "/Library/Python/2.7/site-packages/celery/canvas.py", line 467, in type
    return self._type or self.tasks[0].type.app.tasks['celery.chord']
AttributeError: 'AsyncResult' object has no attribute 'type'

-

@task()
def shutdown(ec2):
    print "shutting down!!!!"
    time.sleep(300)
    return True

c = Ec2()
account_list = Account.objects.all()
chord([change_bio_task.delay() for account in account_list ])(shutdown.s(c))

Answer 1

Chord 接受两个参数：第一个是一个子任务的列表，这个列表叫做“组”；第二个是可选的子任务，用作在列表中的所有任务完成后进行回调。

API 参考中的示例：

res = chord([add.s(2, 2), add.s(4, 4)])(sum_task.s())

在你的代码中，你传递的第一个参数是 AsyncResults 的列表，而不是子任务。这应该是正确的：

chord([change_bio_task.s() for account in account_list ])(shutdown.si(c))

注意我把 shutdown.s(c) 改成了 shutdown.si(c)，这样是不可变的，并且会忽略已经完成的 change_bio_task 返回的结果。