Python中重试模块retrying

在Python开发中，我们经常会遇到一些不稳定的操作，比如网络请求超时、数据库连接失败等。为了提高程序的健壮性，我们需要实现重试机制。retrying是一个非常实用的Python重试库，可以帮助我们轻松实现各种重试场景。

1. 安装

pip install retrying

2. 基本用法

retrying提供了一个retry装饰器，被装饰的函数在运行失败时会自动重试：

import random
from retrying import retry

@retry
def do_something_unreliable():
    if random.randint(0, 10) > 1:
        raise IOError("Broken sauce, everything is hosed!!!111one")
    else:
        return "Awesome sauce!"

print(do_something_unreliable())

3. 配置参数

retry装饰器支持多种配置参数，用于控制重试行为：

3.1 停止策略

• stop_max_attempt_number：最大尝试次数，超过该次数就停止重试
• stop_max_delay：最大延迟时间（毫秒），从函数开始执行到结束的总时间超过此值就停止重试

3.2 等待策略

• wait_fixed：固定等待时间（毫秒）
• wait_random_min和wait_random_max：随机等待时间范围（毫秒）
• wait_exponential_multiplier和wait_exponential_max：指数退避等待策略
  • 计算公式：2^attempt_number * wait_exponential_multiplier
  • 最大值不超过wait_exponential_max
  • 符合exponential backoff算法，可减轻系统压力

3.3 重试条件

• retry_on_exception：根据异常类型决定是否重试
• retry_on_result：根据返回值决定是否重试

4. 高级用法

4.1 根据异常类型重试

from retrying import retry

def retry_if_io_error(exception):
    """当遇到IOError时重试"""
    return isinstance(exception, IOError)

@retry(retry_on_exception=retry_if_io_error, stop_max_attempt_number=3)
def read_a_file():
    """读取文件，遇到IOError时重试最多3次"""
    with open("file.txt", "r") as f:
        return f.read()

# 测试
try:
    content = read_a_file()
    print(f"文件内容: {content}")
except Exception as e:
    print(f"读取失败: {e}")

4.2 根据返回值重试

from retrying import retry

def retry_if_result_none(result):
    """当返回值为None时重试"""
    return result is None

def retry_if_result_error(result):
    """当返回值为错误标记时重试"""
    return result == "error"

@retry(
    retry_on_result=retry_if_result_error,
    stop_max_attempt_number=5,
    wait_fixed=1000  # 每次重试间隔1秒
)
def might_return_error():
    """模拟可能返回错误的函数"""
    import random
    if random.randint(0, 3) > 0:
        print("返回错误，将重试")
        return "error"
    else:
        print("返回成功")
        return "success"

# 测试
result = might_return_error()
print(f"最终结果: {result}")

4.3 组合多种条件

from retrying import retry
import requests

def retry_if_connection_error(exception):
    """当遇到连接错误时重试"""
    return isinstance(exception, (requests.ConnectionError, requests.Timeout))

def retry_if_server_error(result):
    """当服务器返回5xx错误时重试"""
    return hasattr(result, 'status_code') and 500 <= result.status_code < 600

@retry(
    retry_on_exception=retry_if_connection_error,
    retry_on_result=retry_if_server_error,
    stop_max_attempt_number=3,
    wait_exponential_multiplier=1000,  # 指数退避，初始1秒
    wait_exponential_max=10000  # 最大等待时间10秒
)
def fetch_data(url):
    """获取URL数据，遇到连接错误或服务器错误时重试"""
    print(f"尝试获取: {url}")
    response = requests.get(url, timeout=5)
    print(f"响应状态码: {response.status_code}")
    return response

# 测试
try:
    response = fetch_data("https://api.example.com/data")
    print(f"获取成功: {response.json()}")
except Exception as e:
    print(f"获取失败: {e}")

5. 实际应用场景

5.1 网络请求重试

import requests
from retrying import retry

def retry_if_network_error(exception):
    """网络错误重试条件"""
    return isinstance(exception, (
        requests.ConnectionError,
        requests.Timeout,
        requests.RequestException
    ))

@retry(
    retry_on_exception=retry_if_network_error,
    stop_max_attempt_number=3,
    wait_exponential_multiplier=1000,
    wait_exponential_max=5000
)
def get_api_data(url, headers=None):
    """获取API数据，网络错误时自动重试"""
    response = requests.get(url, headers=headers, timeout=10)
    response.raise_for_status()  # 抛出HTTP错误
    return response.json()

# 测试
api_url = "https://api.github.com/users/octocat"
try:
    data = get_api_data(api_url)
    print(f"用户名: {data['login']}")
    print(f"仓库数: {data['public_repos']}")
except Exception as e:
    print(f"API请求失败: {e}")

5.2 数据库操作重试

import psycopg2
from retrying import retry

def retry_if_db_error(exception):
    """数据库错误重试条件"""
    return isinstance(exception, (psycopg2.OperationalError, psycopg2.InterfaceError))

@retry(
    retry_on_exception=retry_if_db_error,
    stop_max_attempt_number=3,
    wait_fixed=2000
)
def execute_db_query(connection_string, query):
    """执行数据库查询，连接错误时重试"""
    conn = psycopg2.connect(connection_string)
    cursor = conn.cursor()
    cursor.execute(query)
    result = cursor.fetchall()
    cursor.close()
    conn.close()
    return result

# 测试
db_conn_str = "host=localhost dbname=mydb user=postgres password=secret"
query = "SELECT * FROM users LIMIT 5"
try:
    results = execute_db_query(db_conn_str, query)
    print(f"查询结果: {results}")
except Exception as e:
    print(f"数据库操作失败: {e}")

5.3 文件操作重试

import os
from retrying import retry

def retry_if_file_error(exception):
    """文件操作错误重试条件"""
    return isinstance(exception, IOError)

@retry(
    retry_on_exception=retry_if_file_error,
    stop_max_attempt_number=3,
    wait_fixed=1000
)
def write_file_with_retry(file_path, content):
    """写入文件，遇到IO错误时重试"""
    with open(file_path, 'w') as f:
        f.write(content)
    print(f"文件写入成功: {file_path}")

# 测试
try:
    write_file_with_retry("output.txt", "Hello, retrying!")
except Exception as e:
    print(f"文件写入失败: {e}")

6. 最佳实践

1. 合理设置重试次数：不要设置过多的重试次数，避免无限重试导致系统资源浪费
2. 使用指数退避策略：对于网络请求，使用指数退避可以减轻服务器压力
3. 明确重试条件：只对可恢复的错误进行重试，如网络超时、临时连接失败等
4. 添加日志：在重试过程中添加日志，便于排查问题
5. 设置超时时间：避免重试时间过长影响整体程序性能
6. 处理最终失败：即使有重试机制，也要处理最终失败的情况

7. 与其他重试库的对比

库	优点	缺点
retrying	简单易用，装饰器形式，配置灵活	功能相对简单，不支持异步
tenacity	更现代，支持异步，API更清晰	相对较新，社区可能不如retrying活跃
backoff	专注于指数退避策略，配置简洁	功能相对单一
自定义重试	完全可控，可根据具体场景定制	代码复杂度高，需要重复实现

8. 注意事项

1. 避免重试幂等操作：只对幂等操作进行重试，避免重复执行导致副作用
2. 注意异常处理：确保重试逻辑不会吞掉所有异常
3. 监控重试频率：如果某个操作频繁重试，可能表明存在更深层次的问题
4. 考虑使用断路器模式：对于频繁失败的操作，考虑使用断路器模式避免雪崩效应

9. 安装替代方案

如果retrying库无法满足需求，可以考虑使用tenacity库，它是retrying的现代替代品：

pip install tenacity

tenacity的使用示例：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def flaky_function():
    import random
    if random.randint(0, 10) > 1:
        raise Exception("Flaky error")
    return "Success!"

print(flaky_function())

10. 参考资料

• retrying GitHub仓库
• tenacity GitHub仓库
• Python重试机制最佳实践
• Exponential Backoff算法详解
• Requests库官方文档