📚 课程目录第4章重试条件控制

4、重试条件控制：retry_condition参数、自定义重试条件、基于异常类型的重试、基于返回值的重试

上一章我们聊了重试延迟和指数退避，那些是「什么时候重试」的问题。这一章我们聊聊「什么情况下才重试」——说白了，就是重试的条件控制。

你想想看，不是所有失败都值得重试的。比如数据库连不上了，重试几次可能就恢复了。但如果是SQL语法写错了，你重试一万次也是白搭。Prefect 给了我们一套很灵活的条件控制机制，让我带你一个个过。

核心思想：重试不是无脑重试，而是有选择地重试。把重试资源花在「有可能恢复」的失败上。

4.1 retry_condition 参数：重试的守门员

retry_condition 是 @task 装饰器里的一个参数。它接收一个可调用对象（函数），这个函数会在任务失败时被调用，决定要不要重试。

它的签名长这样：

def should_retry(task, flow, state, result):
    # task: 当前任务对象
    # flow: 当前流程对象
    # state: 任务失败后的状态
    # result: 任务返回的结果或抛出的异常
    return True  # 返回 True 就重试，False 就不重试

我个人习惯把这个函数单独拎出来写，而不是用 lambda。为什么？因为逻辑复杂了 lambda 根本写不清楚，你想想看。

4.2 基于异常类型的重试：最常见的场景

这是我在项目中用得最多的方式。不同的异常，处理策略完全不同。

举个例子：

from prefect import task, Flow
from prefect.engine.results import PrefectResult
import requests

def retry_on_connection_error(task, flow, state, result):
    """只在连接类异常时重试"""
    if isinstance(result, (requests.ConnectionError, requests.Timeout)):
        return True
    # 如果是 4xx 客户端错误，不重试
    if isinstance(result, requests.HTTPError):
        if result.response.status_code < 500:
            return False
        return True  # 5xx 服务端错误可以重试
    return False

@task(
    max_retries=3,
    retry_delay=10,
    retry_condition=retry_on_connection_error
)
def fetch_data(url):
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()

我的经验：我曾经在一个数据采集项目里，把所有异常都重试了。结果有个 API 返回 400 错误（参数错误），重试了 5 次才放弃，白白浪费了 50 秒。从那以后，我坚持「按异常类型区分重试」。

常见的异常类型重试策略，我整理了一张表：

异常类型	是否重试	原因
`ConnectionError`	✅ 重试	网络抖动，大概率能恢复
`TimeoutError`	✅ 重试	服务端负载高，稍后可能正常
`ValueError`	❌ 不重试	数据格式问题，重试也没用
`FileNotFoundError`	❌ 不重试	文件缺失，重试不会凭空出现
`HTTP 429` (Too Many Requests)	✅ 重试（带延迟）	限流了，等一会再试
`HTTP 5xx`	✅ 重试	服务端临时故障

4.3 基于返回值的重试：不只是看异常

有些任务不会抛异常，但返回的结果「不对劲」。比如返回了空列表、None、或者一个错误码。这时候我们也要能触发重试。

Prefect 的 retry_condition 回调里，result 参数就是任务的返回值。如果任务正常返回，result 就是返回值；如果抛了异常，result 就是异常对象。

看个例子：

def retry_on_empty_result(task, flow, state, result):
    """如果返回空数据，就重试"""
    # 先判断是不是异常
    if isinstance(result, Exception):
        return False  # 异常情况交给其他逻辑处理
    
    # 检查返回值
    if result is None:
        return True
    if isinstance(result, (list, dict)) and len(result) == 0:
        return True
    if isinstance(result, str) and result.strip() == '':
        return True
    
    return False

@task(
    max_retries=2,
    retry_delay=5,
    retry_condition=retry_on_empty_result
)
def query_database(query):
    # 假设这个查询偶尔会返回空结果（但没抛异常）
    result = db.execute(query)
    return result  # 可能返回 None 或空列表

注意：基于返回值重试有个坑——你得确保「空结果」真的是异常情况。如果业务上就是可能查不到数据，那重试反而会引入 bug。我建议在重试条件里加个计数器，或者结合业务状态判断。

4.4 组合条件：更精细的控制

实际项目中，很少只用单一条件。我经常把异常类型和返回值结合起来判断。

比如这样：

def combined_retry_condition(task, flow, state, result):
    """组合条件：异常类型 + 返回值 + 重试次数"""
    # 规则1：连接类异常，最多重试3次
    if isinstance(result, (ConnectionError, TimeoutError)):
        return state.run_count < 3
    
    # 规则2：服务端错误，最多重试5次
    if isinstance(result, HTTPError) and result.response.status_code >= 500:
        return state.run_count < 5
    
    # 规则3：返回空数据，但只在第一次重试
    if not isinstance(result, Exception):
        if result is None or (isinstance(result, list) and len(result) == 0):
            return state.run_count < 2  # 只重试一次
    
    # 其他情况不重试
    return False

嗯，这里要注意 state.run_count 这个属性。它记录了当前任务已经执行了多少次（包括第一次）。用它可以实现「不同阶段不同策略」的逻辑。

4.5 知识体系：重试条件控制全景图

下面这张图帮你理清整个重试条件控制的脉络：

4.6 避坑指南：我踩过的几个坑

讲完了理论，分享几个实战中容易翻车的地方。

坑一：忘记处理异常情况——retry_condition 回调里的 result 可能是异常对象，也可能是正常返回值。如果你只写了 result is None 的判断，当任务抛异常时，这个判断会返回 False，导致异常情况永远不会重试。我建议在回调开头先判断 isinstance(result, Exception)。
坑二：重试条件太宽松——我曾经把「只要不是成功就重试」写进条件里，结果有个任务因为参数错误一直重试，直到耗尽所有重试次数。浪费了资源不说，日志里全是无意义的错误记录。
坑三：忽略重试次数上限——即使条件判断为 True，Prefect 还是会受 max_retries 的限制。但如果你在条件里自己加了次数判断（比如 state.run_count < 5），记得和 max_retries 保持一致，否则逻辑会混乱。

我的调试技巧：在 retry_condition 回调里加个日志打印，把 result 的类型和值、state.run_count 都打出来。这样你能清楚地看到每次重试决策的依据，排查问题特别快。

4.7 小结

这一章我们聊了重试条件控制的几种方式：基于异常类型、基于返回值、以及组合条件。核心就一句话——有选择地重试，别做无用功。