从任务中恢复失败超过 max_retries
Posted
技术标签:
【中文标题】从任务中恢复失败超过 max_retries【英文标题】:Recover from task failed beyond max_retries 【发布时间】:2011-09-23 21:42:43 【问题描述】:我正在尝试异步使用 Web 服务,因为它最多需要 45 秒才能返回。不幸的是,此 Web 服务也有些不可靠,并且可能会引发错误。我已经设置了django-celery
并执行了我的任务,它工作正常,直到任务失败超出max_retries
。
这是我目前所拥有的:
@task(default_retry_delay=5, max_retries=10)
def request(xml):
try:
server = Client('https://www.whatever.net/RealTimeService.asmx?wsdl')
xml = server.service.RunRealTimeXML(
username=settings.WS_USERNAME,
password=settings.WS_PASSWORD,
xml=xml
)
except Exception, e:
result = Result(celery_id=request.request.id, details=e.reason, status="i")
result.save()
try:
return request.retry(exc=e)
except MaxRetriesExceededError, e:
result = Result(celery_id=request.request.id, details="Max Retries Exceeded", status="f")
result.save()
raise
result = Result(celery_id=request.request.id, details=xml, status="s")
result.save()
return result
不幸的是,MaxRetriesExceededError
没有被retry()
抛出,所以我不确定如何处理此任务的失败。 Django 已经将 html 返回给客户端,我正在通过 AJAX 检查Result
的内容,这永远不会完全失败f
状态。
所以问题是:当 Celery 任务超过 max_retries
时,如何更新我的数据库?
【问题讨论】:
【参考方案1】:你可以重写celery任务类的after_return方法,这个方法在任务执行后无论ret状态是什么(SUCCESS,FAILED,RETRY)都会被调用
class MyTask(celery.task.Task)
def run(self, xml, **kwargs)
#Your stuffs here
def after_return(self, status, retval, task_id, args, kwargs, einfo=None):
if self.max_retries == int(kwargs['task_retries']):
#If max retries are equals to task retries do something
if status == "FAILURE":
#You can do also something if the tasks fail instead of check the retries
http://readthedocs.org/docs/celery/en/latest/reference/celery.task.base.html#celery.task.base.BaseTask.after_return
http://celery.readthedocs.org/en/latest/reference/celery.app.task.html?highlight=after_return#celery.app.task.Task.after_return
【讨论】:
由于链接现在显然已过时,here is a new one【参考方案2】:对于 Celery 2.3.2 版,这种方法对我来说效果很好:
class MyTask(celery.task.Task):
abstract = True
def after_return(self, status, retval, task_id, args, kwargs, einfo):
if self.max_retries == self.request.retries:
#If max retries is equal to task retries do something
@task(base=MyTask, default_retry_delay=5, max_retries=10)
def request(xml):
#Your stuff here
【讨论】:
【参考方案3】:我现在只是这样做,省去了子类化 Task 的工作,并且很容易理解。
# auto-retry with delay as defined below. After that, hook is disabled.
@celery.shared_task(bind=True, max_retries=5, default_retry_delay=300)
def post_data(self, hook_object_id, url, event, payload):
headers = 'Content-type': 'application/json'
try:
r = requests.post(url, data=payload, headers=headers)
r.raise_for_status()
except requests.exceptions.RequestException as e:
if self.request.retries >= self.max_retries:
log.warning("Auto-deactivating webhook %s for event %s", hook_object_id, event)
Webhook.objects.filter(object_id=hook_object_id).update(active=False)
return False
raise self.retry(exc=e)
return True
【讨论】:
谢谢!和你的解决方案一起去了。我还添加了throw=False
。我已经尝试过使用MaxRetriesExceededError
的其他解决方案,当为一组任务调用.join()
时,该过程会卡住。【参考方案4】:
问题是 celery 在达到重试限制时试图重新引发您传入的异常。重新加注的代码在这里:https://github.com/celery/celery/blob/v3.1.20/celery/app/task.py#L673-L681
解决这个问题的最简单方法是根本不让 celery 管理您的异常:
@task(max_retries=10)
def mytask():
try:
do_the_thing()
except Exception as e:
try:
mytask.retry()
except MaxRetriesExceededError:
do_something_to_handle_the_error()
logger.exception(e)
【讨论】:
这是解决原始问题的正确方法。 不起作用。获取RuntimeError: Never call result.get() within a task!
以上是关于从任务中恢复失败超过 max_retries的主要内容,如果未能解决你的问题,请参考以下文章