我遇到了一个非常奇怪的芹菜问题:
有一系列任务,其中一个给出异常并重试几次
chain = (err.si(1) | err.si(2))
result = chain.apply_async()
result.state
result.get()
这是任务的代码:
@celery.task(base=MyTask)
def err(x):
try:
if x < 3:
raise Exception
else:
return x+1
except Exception as exp:
print "retrying"
raise err.retry(args=[x],exc=exp,countdown=5,max_retries=3)
问题是,虽然链中的任务给出了一个异常,但是 result.state 一直是“PENDING”并且 .get() 只是冻结了。
如果它达到最大重试值,我试图使任务失败:
class MyTask(celery.Task):
abstract = True
def after_return(self, status, retval, task_id, args, kwargs, einfo):
if self.max_retries == self.request.retries:
self.state = states.FAILURE
但是,尽管单独执行的任务被标记为 FAILED,但在链中执行会给出相同的结果 - PENDING & Freezed get。
我预计一旦它的任何任务失败并且 .get 结果应该产生从任务抛出的异常,该链将失败。
_ UPDATE _ apply_async 给出的堆栈跟踪,其中 ALWAYS_EAGER=True
result = chain.apply_async()
Exception
Traceback (most recent call last)
<ipython-input-4-81202b369b5f> in <module>()
----> 1 result = chain.apply_async()
lib/python2.7/site-packages/celery/canvas.pyc in apply_async(self, args, kwargs, **options)
147 # For callbacks: extra args are prepended to the stored args.
148 args, kwargs, options = self._merge(args, kwargs, options)
--> 149 return self.type.apply_async(args, kwargs, **options)
150
151 def append_to_list_option(self, key, value):
/lib/python2.7/site-packages/celery/app/builtins.pyc in apply_async(self, args, kwargs, group_id, chord, task_id, **options)
232 task_id=None, **options):
233 if self.app.conf.CELERY_ALWAYS_EAGER:
--> 234 return self.apply(args, kwargs, **options)
235 options.pop('publisher', None)
236 tasks, results = self.prepare_steps(args, kwargs['tasks'])
lib/python2.7/site-packages/celery/app/builtins.pyc in apply(self, args, kwargs, subtask, **options)
249 last, fargs = None, args # fargs passed to first task only
250 for task in kwargs['tasks']:
--> 251 res = subtask(task).clone(fargs).apply(last and (last.get(), ))
252 res.parent, last, fargs = last, res, None
253 return last
lib/python2.7/site-packages/celery/result.pyc in get(self, timeout, propagate, **kwargs)
677 elif self.state in states.PROPAGATE_STATES:
678 if propagate:
--> 679 raise self.result
680 return self.result
681 wait = get
Exception: