如何为Celery任务的不同异常类型设置独立重试次数？

当前你的代码依赖Celery全局的retries计数，导致不同异常的重试次数互相干扰——比如先触发6次ConnectionError后遇到ReadTimeout，全局重试次数已到6，超过ReadTimeout设置的5次上限，任务直接终止，无法实现「每种异常单独计数重试」的需求。

要解决这个问题，我们可以手动维护每种异常的重试计数，通过Celery任务的kwargs传递和存储这些计数，具体实现如下：

from celery import Celery
import requests

app = Celery('tasks', broker='pyamqp://guest@localhost//')

@app.task(bind=True, serializer='json')
def create_user(self, client, **kwargs):
    # 从kwargs读取各异常的重试计数，默认初始为0
    conn_retry_count = kwargs.get('conn_retry_count', 0)
    timeout_retry_count = kwargs.get('timeout_retry_count', 0)
    
    try:
        data = client.get('users/')
        return data
    except requests.ConnectionError as e:
        conn_retry_count += 1
        # 判断是否达到该异常的重试上限
        if conn_retry_count >= 10:
            raise e  # 超过次数，终止任务
        # 更新kwargs中的计数，传递给下一次重试
        self.request.kwargs['conn_retry_count'] = conn_retry_count
        raise self.retry(exc=e, kwargs=self.request.kwargs) from e
    except requests.ReadTimeout as e:
        timeout_retry_count += 1
        if timeout_retry_count >= 5:
            raise e
        self.request.kwargs['timeout_retry_count'] = timeout_retry_count
        raise self.retry(exc=e, kwargs=self.request.kwargs) from e

核心逻辑说明： #

• 放弃依赖Celery全局的self.request.retries，改用kwargs单独存储每种异常的重试次数，实现不同异常计数完全隔离
• 每次捕获对应异常时，先递增对应计数，判断是否达到预设上限：达到则直接抛出异常终止任务，未达到则更新kwargs并触发重试
• Celery重试时会自动传递kwargs中的参数，因此每次重试都能拿到之前的计数

这种方式完全满足需求：ConnectionError最多重试10次，ReadTimeout最多重试5次，两种异常的重试次数互不影响。

内容的提问来源于stack exchange，提问作者Shamsiddin Parpiev