celery任务失败重试

方式一：

示例：

@app.task(bind=True)
def send_twitter_status(self, oauth, tweet):
    try:
        twitter = Twitter(oauth)
        twitter.update_status(tweet)
    except (Twitter.FailWhaleError, Twitter.LoginError) as exc:
        raise self.retry(exc=exc)

retry的参数可以有：

exc：指定抛出的异常
throw：重试时是否通知worker是重试任务
eta：指定重试的时间／日期
countdown：在多久之后重试（每多少秒重试一次）
max_retries：最大重试次数

bing=True后，task对象会作为第一个参数自动传入，可以使用任务对象的属性。例如：
self.request.retries：当前重试的次数
self.request还有以下属性：

name	desc
id	The unique id of the executing task
group	The unique id a group, if this task is a member.
chord	The unique id of the chord this task belongs to (if the task is part of the header).
args	Positional arguments.
kwargs	Keyword arguments.
retries	How many times the current task has been retried. An integer starting at 0.
is_eager	Set to True if the task is executed locally in the client, and not by a worker.
eta	The original ETA of the task (if any). This is in UTC time (depending on the CELERY_ENABLE_UTC setting).
expires	The original expiry time of the task (if any). This is in UTC time (depending on the CELERY_ENABLE_UTC setting).
logfile	The file the worker logs to. See Logging.
loglevel	The current log level used.
hostname	Hostname of the worker instance executing the task.
delivery_info	Additional message delivery information. This is a mapping containing the exchange and routing key used to deliver this task. Used by e.g. retry() to resend the task to the same destination queue. Availability of keys in this dict depends on the message broker used.
called_directly	This flag is set to true if the task was not executed by the worker.
callbacks	A list of subtasks to be called if this task returns successfully.
errback	A list of subtasks to be called if this task fails.
utc	Set to true the caller has utc enabled (CELERY_ENABLE_UTC).

方式二：

@app.task(autoretry_for=(ReadTimeout,), retry_kwargs={'max_retries': 3, 'countdown': 5})
def test_func():
    viewutils.test_func()

在autoretry_for中添加要自动重试的异常，如果所有异常都需要重试可以写Exception，retry_kwargs中添加重试的参数。

方式三：

重写Task类

from celery import Task

class DebugTask(Task):
    abstract = True

    def after_return(self, *args, **kwargs):
        print('Task returned: {0!r}'.format(self.request)


@app.task(base=DebugTask)
def add(x, y):
    return x + y

Handlers

after_return(self, status, retval, task_id, args, kwargs, einfo)
Handler called after the task returns.
参数:
- status – Current task state.
- retval – Task return value/exception.
- task_id – Unique id of the task.
- args – Original arguments for the task that returned.
- kwargs – Original keyword arguments for the task that returned.
- einfo – ExceptionInfo instance, containing the traceback (if any).
The return value of this handler is ignored.
on_failure(self, exc, task_id, args, kwargs, einfo)
This is run by the worker when the task fails.
参数:
- exc – The exception raised by the task.
- task_id – Unique id of the failed task.
- args – Original arguments for the task that failed.
- kwargs – Original keyword arguments for the task that failed.
- einfo – ExceptionInfo instance, containing the traceback.
The return value of this handler is ignored.
on_retry(self, exc, task_id, args, kwargs, einfo)
This is run by the worker when the task is to be retried.
参数:
- exc – The exception sent to retry().
- task_id – Unique id of the retried task.
- args – Original arguments for the retried task.
- kwargs – Original keyword arguments for the retried task.
- einfo – ExceptionInfo instance, containing the traceback.
The return value of this handler is ignored.
on_success(self, retval, task_id, args, kwargs)
Run by the worker if the task executes successfully.
参数:
- retval – The return value of the task.
- task_id – Unique id of the executed task.
- args – Original arguments for the executed task.
- kwargs – Original keyword arguments for the executed task.
The return value of this handler is ignored.

原文链接：https://blog.csdn.net/yezi1993/article/details/89455716