使用 autoretry_for 正确处理 Celery 异常
Correctly handle Celery Exceptions using autoretry_for
我有一个用 autoretry_for 修饰的 celery 任务，这样在已知异常的情况下，它会重试该任务。此处为虚拟版本：
class ExpectedException(Exception):
pass
@app.task(autoretry_for=(ExpectedException,), retry_kwargs={'max_retries': 2, 'countdown': 1})
def decorated_autoretry():
logging.info(
"Attempt: {attempt} of {attempts}".format(
attempt=decorated_autoretry.request.retries, attempts=decorated_autoretry.max_retries
)
)
raise ExpectedException
当 运行 给出以下输出时：
[2018-01-30 12:17:31,899: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a]
[2018-01-30 12:17:31,900: INFO/ForkPoolWorker-1] Attempt: 1 of 3
[2018-01-30 12:17:31,915: INFO/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a] retry: Retry in 1s: ExpectedException()
[2018-01-30 12:17:31,915: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a] ETA:[2018-01-30 12:17:32.901955+00:00]
[2018-01-30 12:17:33,024: INFO/ForkPoolWorker-2] Attempt: 2 of 3
[2018-01-30 12:17:33,072: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a] ETA:[2018-01-30 12:17:34.029462+00:00]
[2018-01-30 12:17:33,072: INFO/ForkPoolWorker-2] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a] retry: Retry in 1s: ExpectedException()
[2018-01-30 12:17:34,033: INFO/ForkPoolWorker-1] Attempt: 3 of 3
[2018-01-30 12:17:34,037: ERROR/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a] raised unexpected: ExpectedException()
Traceback (most recent call last):
File "/home/greg/gel/interpretation/.env/local/lib/python2.7/site-packages/celery/app/trace.py", line 374, in trace_task
R = retval = fun(*args, **kwargs)
File "/home/greg/gel/interpretation/.env/local/lib/python2.7/site-packages/celery/app/trace.py", line 629, in __protected_call__
return self.run(*args, **kwargs)
File "/home/greg/gel/interpretation/.env/local/lib/python2.7/site-packages/celery/app/base.py", line 474, in run
raise task.retry(exc=exc, **retry_kwargs)
File "/home/greg/gel/interpretation/.env/local/lib/python2.7/site-packages/celery/app/task.py", line 669, in retry
raise_with_context(exc)
File "/home/greg/gel/interpretation/.env/local/lib/python2.7/site-packages/celery/app/base.py", line 472, in run
return task._orig_run(*args, **kwargs)
File "/home/greg/gel/interpretation/interpretationAPI/tasks/util_tasks.py", line 25, in decorated_autoretry
raise ExpectedException
ExpectedException
在最后一次尝试中引发了 ExpectedException，但 Celery 认为这是意外的。
我可以在这里明确处理异常：
@app.task
def explicit_autoretry():
logging.info(
"Attempt: {attempt} of {attempts}".format(
attempt=explicit_autoretry.request.retries+1, attempts=explicit_autoretry.max_retries
)
)
try:
raise ExpectedException
except ExpectedException as e:
logging.info(msg="Received exception of type: {e_type}".format(e_type=type(e)))
try:
explicit_autoretry.retry(countdown=1)
except MaxRetriesExceededError as e:
logging.info(msg="Received exception of type: {e_type}".format(e_type=type(e)))
当 运行 给出以下输出时：
[2018-01-30 12:19:45,284: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9]
[2018-01-30 12:19:45,287: INFO/ForkPoolWorker-3] Attempt: 1 of 3
[2018-01-30 12:19:45,288: INFO/ForkPoolWorker-3] Received exception of type: <class 'interpretationAPI.tasks.util_tasks.ExpectedException'>
[2018-01-30 12:19:45,301: INFO/ForkPoolWorker-3] Task interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] retry: Retry in 1s
[2018-01-30 12:19:45,301: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] ETA:[2018-01-30 12:19:46.288992+00:00]
[2018-01-30 12:19:47,790: INFO/ForkPoolWorker-1] Attempt: 2 of 3
[2018-01-30 12:19:47,792: INFO/ForkPoolWorker-1] Received exception of type: <class 'interpretationAPI.tasks.util_tasks.ExpectedException'>
[2018-01-30 12:19:47,839: INFO/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] retry: Retry in 1s
[2018-01-30 12:19:47,839: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] ETA:[2018-01-30 12:19:48.796492+00:00]
[2018-01-30 12:19:49,789: INFO/ForkPoolWorker-3] Attempt: 3 of 3
[2018-01-30 12:19:49,789: INFO/ForkPoolWorker-3] Received exception of type: <class 'interpretationAPI.tasks.util_tasks.ExpectedException'>
[2018-01-30 12:19:49,791: INFO/ForkPoolWorker-3] Task interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] retry: Retry in 1s
[2018-01-30 12:19:49,791: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] ETA:[2018-01-30 12:19:50.790244+00:00]
[2018-01-30 12:19:51,791: INFO/ForkPoolWorker-1] Attempt: 4 of 3
[2018-01-30 12:19:51,791: INFO/ForkPoolWorker-1] Received exception of type: <class 'interpretationAPI.tasks.util_tasks.ExpectedException'>
[2018-01-30 12:19:51,791: INFO/ForkPoolWorker-1] Received exception of type: <class 'celery.exceptions.MaxRetriesExceededError'>
[2018-01-30 12:19:51,791: INFO/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] succeeded in 0.00055760199939s: None
正确处理 ExpectedException 的每个实例和最终的 MaxRetriesExceededError，允许优雅的错误处理。
有没有一种在使用 autoretry_for 装饰器时进行此类错误处理的方法，从而最大限度地减少显式错误处理的数量？我也尝试过使用 on_failure 处理程序但无济于事。
将throws=(ExpectedException,)添加到装饰器。
修改上面的例子：
@app.task(throws=(ExpectedException,), autoretry_for=(ExpectedException,), retry_kwargs={'max_retries': 2, 'countdown': 1})
def decorated_autoretry():
logging.info(
"Attempt: {attempt} of {attempts}".format(
attempt=decorated_autoretry.request.retries, attempts=decorated_autoretry.max_retries
)
)
raise ExpectedException
将运行给予：
[2018-01-30 12:43:48,684: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0]
[2018-01-30 12:43:48,685: INFO/ForkPoolWorker-1] Attempt: 1 of 3
[2018-01-30 12:43:48,687: INFO/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0] retry: Retry in 1s: ExpectedException()
[2018-01-30 12:43:48,688: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0] ETA:[2018-01-30 12:43:49.685804+00:00]
[2018-01-30 12:43:51,641: INFO/ForkPoolWorker-2] Attempt: 2 of 3
[2018-01-30 12:43:51,643: INFO/ForkPoolWorker-2] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0] retry: Retry in 1s: ExpectedException()
[2018-01-30 12:43:51,643: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0] ETA:[2018-01-30 12:43:52.642046+00:00]
[2018-01-30 12:43:53,459: INFO/ForkPoolWorker-1] Attempt: 3 of 3
[2018-01-30 12:43:53,461: INFO/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0] raised expected: ExpectedException()
并且会正常退出，正如我们从消息的 raised expected: ExpectedException() 部分看到的那样。
我有一个用 autoretry_for 修饰的 celery 任务，这样在已知异常的情况下，它会重试该任务。此处为虚拟版本：
class ExpectedException(Exception):
pass
@app.task(autoretry_for=(ExpectedException,), retry_kwargs={'max_retries': 2, 'countdown': 1})
def decorated_autoretry():
logging.info(
"Attempt: {attempt} of {attempts}".format(
attempt=decorated_autoretry.request.retries, attempts=decorated_autoretry.max_retries
)
)
raise ExpectedException
当 运行 给出以下输出时：
[2018-01-30 12:17:31,899: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a]
[2018-01-30 12:17:31,900: INFO/ForkPoolWorker-1] Attempt: 1 of 3
[2018-01-30 12:17:31,915: INFO/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a] retry: Retry in 1s: ExpectedException()
[2018-01-30 12:17:31,915: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a] ETA:[2018-01-30 12:17:32.901955+00:00]
[2018-01-30 12:17:33,024: INFO/ForkPoolWorker-2] Attempt: 2 of 3
[2018-01-30 12:17:33,072: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a] ETA:[2018-01-30 12:17:34.029462+00:00]
[2018-01-30 12:17:33,072: INFO/ForkPoolWorker-2] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a] retry: Retry in 1s: ExpectedException()
[2018-01-30 12:17:34,033: INFO/ForkPoolWorker-1] Attempt: 3 of 3
[2018-01-30 12:17:34,037: ERROR/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[9e00f56d-fb90-46db-a735-678fd0b4cb5a] raised unexpected: ExpectedException()
Traceback (most recent call last):
File "/home/greg/gel/interpretation/.env/local/lib/python2.7/site-packages/celery/app/trace.py", line 374, in trace_task
R = retval = fun(*args, **kwargs)
File "/home/greg/gel/interpretation/.env/local/lib/python2.7/site-packages/celery/app/trace.py", line 629, in __protected_call__
return self.run(*args, **kwargs)
File "/home/greg/gel/interpretation/.env/local/lib/python2.7/site-packages/celery/app/base.py", line 474, in run
raise task.retry(exc=exc, **retry_kwargs)
File "/home/greg/gel/interpretation/.env/local/lib/python2.7/site-packages/celery/app/task.py", line 669, in retry
raise_with_context(exc)
File "/home/greg/gel/interpretation/.env/local/lib/python2.7/site-packages/celery/app/base.py", line 472, in run
return task._orig_run(*args, **kwargs)
File "/home/greg/gel/interpretation/interpretationAPI/tasks/util_tasks.py", line 25, in decorated_autoretry
raise ExpectedException
ExpectedException
在最后一次尝试中引发了 ExpectedException，但 Celery 认为这是意外的。
我可以在这里明确处理异常：
@app.task
def explicit_autoretry():
logging.info(
"Attempt: {attempt} of {attempts}".format(
attempt=explicit_autoretry.request.retries+1, attempts=explicit_autoretry.max_retries
)
)
try:
raise ExpectedException
except ExpectedException as e:
logging.info(msg="Received exception of type: {e_type}".format(e_type=type(e)))
try:
explicit_autoretry.retry(countdown=1)
except MaxRetriesExceededError as e:
logging.info(msg="Received exception of type: {e_type}".format(e_type=type(e)))
当 运行 给出以下输出时：
[2018-01-30 12:19:45,284: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9]
[2018-01-30 12:19:45,287: INFO/ForkPoolWorker-3] Attempt: 1 of 3
[2018-01-30 12:19:45,288: INFO/ForkPoolWorker-3] Received exception of type: <class 'interpretationAPI.tasks.util_tasks.ExpectedException'>
[2018-01-30 12:19:45,301: INFO/ForkPoolWorker-3] Task interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] retry: Retry in 1s
[2018-01-30 12:19:45,301: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] ETA:[2018-01-30 12:19:46.288992+00:00]
[2018-01-30 12:19:47,790: INFO/ForkPoolWorker-1] Attempt: 2 of 3
[2018-01-30 12:19:47,792: INFO/ForkPoolWorker-1] Received exception of type: <class 'interpretationAPI.tasks.util_tasks.ExpectedException'>
[2018-01-30 12:19:47,839: INFO/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] retry: Retry in 1s
[2018-01-30 12:19:47,839: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] ETA:[2018-01-30 12:19:48.796492+00:00]
[2018-01-30 12:19:49,789: INFO/ForkPoolWorker-3] Attempt: 3 of 3
[2018-01-30 12:19:49,789: INFO/ForkPoolWorker-3] Received exception of type: <class 'interpretationAPI.tasks.util_tasks.ExpectedException'>
[2018-01-30 12:19:49,791: INFO/ForkPoolWorker-3] Task interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] retry: Retry in 1s
[2018-01-30 12:19:49,791: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] ETA:[2018-01-30 12:19:50.790244+00:00]
[2018-01-30 12:19:51,791: INFO/ForkPoolWorker-1] Attempt: 4 of 3
[2018-01-30 12:19:51,791: INFO/ForkPoolWorker-1] Received exception of type: <class 'interpretationAPI.tasks.util_tasks.ExpectedException'>
[2018-01-30 12:19:51,791: INFO/ForkPoolWorker-1] Received exception of type: <class 'celery.exceptions.MaxRetriesExceededError'>
[2018-01-30 12:19:51,791: INFO/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.explicit_autoretry[b1f29107-ac73-401d-b0c3-636e91661ee9] succeeded in 0.00055760199939s: None
正确处理 ExpectedException 的每个实例和最终的 MaxRetriesExceededError，允许优雅的错误处理。
有没有一种在使用 autoretry_for 装饰器时进行此类错误处理的方法，从而最大限度地减少显式错误处理的数量？我也尝试过使用 on_failure 处理程序但无济于事。
将throws=(ExpectedException,)添加到装饰器。
修改上面的例子：
@app.task(throws=(ExpectedException,), autoretry_for=(ExpectedException,), retry_kwargs={'max_retries': 2, 'countdown': 1})
def decorated_autoretry():
logging.info(
"Attempt: {attempt} of {attempts}".format(
attempt=decorated_autoretry.request.retries, attempts=decorated_autoretry.max_retries
)
)
raise ExpectedException
将运行给予：
[2018-01-30 12:43:48,684: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0]
[2018-01-30 12:43:48,685: INFO/ForkPoolWorker-1] Attempt: 1 of 3
[2018-01-30 12:43:48,687: INFO/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0] retry: Retry in 1s: ExpectedException()
[2018-01-30 12:43:48,688: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0] ETA:[2018-01-30 12:43:49.685804+00:00]
[2018-01-30 12:43:51,641: INFO/ForkPoolWorker-2] Attempt: 2 of 3
[2018-01-30 12:43:51,643: INFO/ForkPoolWorker-2] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0] retry: Retry in 1s: ExpectedException()
[2018-01-30 12:43:51,643: INFO/MainProcess] Received task: interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0] ETA:[2018-01-30 12:43:52.642046+00:00]
[2018-01-30 12:43:53,459: INFO/ForkPoolWorker-1] Attempt: 3 of 3
[2018-01-30 12:43:53,461: INFO/ForkPoolWorker-1] Task interpretationAPI.tasks.util_tasks.decorated_autoretry[34de3cb0-9fec-4bc3-a900-c0eb5d8eb5b0] raised expected: ExpectedException()
并且会正常退出，正如我们从消息的 raised expected: ExpectedException() 部分看到的那样。