我有一个运行一些任务的芹菜链。 每个任务会失败，且重试。 请参阅下面一个简单的例子：
from celery import task
@task(ignore_result=True)
def add(x, y, fail=True):
try:
if fail:
raise Exception('Ugly exception.')
print '%d + %d = %d' % (x, y, x+y)
except Exception as e:
raise add.retry(args=(x, y, False), exc=e, countdown=10)
@task(ignore_result=True)
def mul(x, y):
print '%d * %d = %d' % (x, y, x*y)
和链条：
from celery.canvas import chain
chain(add.si(1, 2), mul.si(3, 4)).apply_async()
运行两个任务（并假设没有失败），你会得到/看印刷：
1 + 2 = 3
3 * 4 = 12
然而，当添加任务失败的第一次，并在随后的重试调用成功，在链中的剩余任务不运行，即添加任务失败，链中的所有其他任务都没有几秒钟后运行，并且，添加任务再次运行，并成功和任务链中的其余部分（在这种情况下mul.si（3,4））不运行。
芹菜是否提供了一种方式来继续从失败，起任务失败链？ 如果不是，将是实现这个和确保该命令所指明运行链的任务，只有以前的任务后，已经成功运行了。即使任务重试几次，最好的办法？
注1：这个问题可以通过做来解决
add.delay(1, 2).get()
mul.delay(3, 4).get()
但我想了解为何链不失败的任务。
你发现了一个bug :)
在固定https://github.com/celery/celery/commit/b2b9d922fdaed5571cf685249bdc46f28acacde3将是3.0.4部分。
我也有兴趣了解为何链不失败的任务。
我挖一些芹菜代码，以及我发现到目前为止是：
实施happends在app.builtins.py
@shared_task
def add_chain_task(app):
from celery.canvas import chord, group, maybe_subtask
_app = app
class Chain(app.Task):
app = _app
name = 'celery.chain'
accept_magic_kwargs = False
def prepare_steps(self, args, tasks):
steps = deque(tasks)
next_step = prev_task = prev_res = None
tasks, results = [], []
i = 0
while steps:
# First task get partial args from chain.
task = maybe_subtask(steps.popleft())
task = task.clone() if i else task.clone(args)
i += 1
tid = task.options.get('task_id')
if tid is None:
tid = task.options['task_id'] = uuid()
res = task.type.AsyncResult(tid)
# automatically upgrade group(..) | s to chord(group, s)
if isinstance(task, group):
try:
next_step = steps.popleft()
except IndexError:
next_step = None
if next_step is not None:
task = chord(task, body=next_step, task_id=tid)
if prev_task:
# link previous task to this task.
prev_task.link(task)
# set the results parent attribute.
res.parent = prev_res
results.append(res)
tasks.append(task)
prev_task, prev_res = task, res
return tasks, results
def apply_async(self, args=(), kwargs={}, group_id=None, chord=None,
task_id=None, **options):
if self.app.conf.CELERY_ALWAYS_EAGER:
return self.apply(args, kwargs, **options)
options.pop('publisher', None)
tasks, results = self.prepare_steps(args, kwargs['tasks'])
result = results[-1]
if group_id:
tasks[-1].set(group_id=group_id)
if chord:
tasks[-1].set(chord=chord)
if task_id:
tasks[-1].set(task_id=task_id)
result = tasks[-1].type.AsyncResult(task_id)
tasks[0].apply_async()
return result
def apply(self, args=(), kwargs={}, **options):
tasks = [maybe_subtask(task).clone() for task in kwargs['tasks']]
res = prev = None
for task in tasks:
res = task.apply((prev.get(), ) if prev else ())
res.parent, prev = prev, res
return res
return Chain
你可以看到，在年底prepare_steps prev_task链接到下一个任务。 当prev_task失败接下来的任务就是不叫。
我与添加从上一个任务link_error到下一个测试：
if prev_task:
# link and link_error previous task to this task.
prev_task.link(task)
prev_task.link_error(task)
# set the results parent attribute.
res.parent = prev_res
不过，接下来的任务必须照顾这两种情况下（也许，当它配置为不可变的，如不接受更多的参数除外）。
我想链可以通过允许一些语法喜欢这个支持：
c = chain(t1, (t2, t1e), (t3, t2e))
意思是：
t1 link到t2和link_error到t1e
t2 link到t3和link_error到t2e
