我有这种情况:
tasks.py
@task
def add(a,b):
return a+b
@task
def other():
chunks = [1,1,1,1] # dummy data
for index in range(3):
# wait for each group to finish then continue to the next one
res = group( add.s(i,i) for i in chunks ).apply_async()
# sleep for 1 second if group is not ready
while not res.get():
time.sleep(1)
这会导致在等待任务组完成时出现死锁吗？即使在只有 1 个 celery worker 的理论情况下？
请您参考如下方法:
您正在等待 other 任务中的 group 任务的结果。因此，即使只有一个 celery worker ，它也可能导致死锁。
让一个任务等待另一个任务的结果是非常低效的，如果工作池耗尽甚至可能导致死锁。
注意:这只是在 Celery 3.1 中发出警告。但从 Celery 3.2 开始，它将引发异常。
所以，最好让你的设计异步。您可以通过简单的修改来实现。
@task
def other():
chunks = [1, 1, 1, 1]
my_tasks = []
for i in range(3):
# delay is shorthand for apply_async.
# using si to make signature immutable,so that its arguments don't change
group_task = group(add.si(i, i) for i in chunks).delay()
# here instead of executing them immediately, lets chain them
my_tasks.append(group_task)
from celery import chain
chain(my_tasks)
