wehub-skill-sync
196 lines
5.6 KiB
Markdown
196 lines
5.6 KiB
Markdown
---
|
||
name: python-resilience
|
||
description: Python 弹性模式，包括自动重试、指数退避、超时和容错装饰器。在需要添加重试逻辑、实现超时、构建容错服务或处理瞬时故障时使用。
|
||
---
|
||
|
||
# Python 弹性模式
|
||
|
||
构建容错的 Python 应用程序，优雅地处理瞬时故障、网络问题和服务中断。当依赖项不可靠时，弹性模式能让系统保持运行。
|
||
|
||
## 何时使用本技能
|
||
|
||
- 为外部服务调用添加重试逻辑
|
||
- 为网络操作实现超时
|
||
- 构建容错的微服务
|
||
- 处理速率限制和背压
|
||
- 创建基础设施装饰器
|
||
- 设计断路器
|
||
|
||
## 核心概念
|
||
|
||
### 1. 瞬时故障与永久故障
|
||
|
||
重试瞬时错误（网络超时、临时服务问题）。不要重试永久错误（无效凭据、错误请求）。
|
||
|
||
### 2. 指数退避
|
||
|
||
增加重试之间的等待时间，避免让正在恢复的服务不堪重负。
|
||
|
||
### 3. 抖动
|
||
|
||
为退避算法添加随机性，防止多个客户端同时重试时产生惊群效应。
|
||
|
||
### 4. 有界重试
|
||
|
||
同时限制重试次数和总持续时间，防止无限重试循环。
|
||
|
||
## 快速开始
|
||
|
||
```python
|
||
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
|
||
|
||
@retry(
|
||
stop=stop_after_attempt(3),
|
||
wait=wait_exponential_jitter(initial=1, max=10),
|
||
)
|
||
def call_external_service(request: dict) -> dict:
|
||
return httpx.post("https://api.example.com", json=request).json()
|
||
```
|
||
|
||
## 基础模式
|
||
|
||
### 模式 1：使用 Tenacity 的基础重试
|
||
|
||
使用 `tenacity` 库实现生产级重试逻辑。对于更简单的场景，可考虑内置重试功能或轻量级自定义实现。
|
||
|
||
```python
|
||
from tenacity import (
|
||
retry,
|
||
stop_after_attempt,
|
||
stop_after_delay,
|
||
wait_exponential_jitter,
|
||
retry_if_exception_type,
|
||
)
|
||
|
||
TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError)
|
||
|
||
@retry(
|
||
retry=retry_if_exception_type(TRANSIENT_ERRORS),
|
||
stop=stop_after_attempt(5) | stop_after_delay(60),
|
||
wait=wait_exponential_jitter(initial=1, max=30),
|
||
)
|
||
def fetch_data(url: str) -> dict:
|
||
"""在瞬时故障时自动重试获取数据。"""
|
||
response = httpx.get(url, timeout=30)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
```
|
||
|
||
### 模式 2：仅重试适当的错误
|
||
|
||
白名单特定的瞬时异常。绝不重试：
|
||
|
||
- `ValueError`、`TypeError`——这些是代码缺陷，不是瞬时问题
|
||
- `AuthenticationError`——无效凭据不会自动变有效
|
||
- HTTP 4xx 错误（429 除外）——客户端错误是永久性的
|
||
|
||
```python
|
||
from tenacity import retry, retry_if_exception_type
|
||
import httpx
|
||
|
||
# 定义哪些是可重试的
|
||
RETRYABLE_EXCEPTIONS = (
|
||
ConnectionError,
|
||
TimeoutError,
|
||
httpx.ConnectTimeout,
|
||
httpx.ReadTimeout,
|
||
)
|
||
|
||
@retry(
|
||
retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS),
|
||
stop=stop_after_attempt(3),
|
||
wait=wait_exponential_jitter(initial=1, max=10),
|
||
)
|
||
def resilient_api_call(endpoint: str) -> dict:
|
||
"""在出现网络问题时带重试的 API 调用。"""
|
||
return httpx.get(endpoint, timeout=10).json()
|
||
```
|
||
|
||
### 模式 3：HTTP 状态码重试
|
||
|
||
重试表示瞬时问题的特定 HTTP 状态码。
|
||
|
||
```python
|
||
from tenacity import retry, retry_if_result, stop_after_attempt
|
||
import httpx
|
||
|
||
RETRY_STATUS_CODES = {429, 502, 503, 504}
|
||
|
||
def should_retry_response(response: httpx.Response) -> bool:
|
||
"""检查响应是否表示可重试的错误。"""
|
||
return response.status_code in RETRY_STATUS_CODES
|
||
|
||
@retry(
|
||
retry=retry_if_result(should_retry_response),
|
||
stop=stop_after_attempt(3),
|
||
wait=wait_exponential_jitter(initial=1, max=10),
|
||
)
|
||
def http_request(method: str, url: str, **kwargs) -> httpx.Response:
|
||
"""在瞬时状态码上带重试的 HTTP 请求。"""
|
||
return httpx.request(method, url, timeout=30, **kwargs)
|
||
```
|
||
|
||
### 模式 4：异常与状态码组合重试
|
||
|
||
同时处理网络异常和 HTTP 状态码。
|
||
|
||
```python
|
||
from tenacity import (
|
||
retry,
|
||
retry_if_exception_type,
|
||
retry_if_result,
|
||
stop_after_attempt,
|
||
wait_exponential_jitter,
|
||
before_sleep_log,
|
||
)
|
||
import logging
|
||
import httpx
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
TRANSIENT_EXCEPTIONS = (
|
||
ConnectionError,
|
||
TimeoutError,
|
||
httpx.ConnectError,
|
||
httpx.ReadTimeout,
|
||
)
|
||
RETRY_STATUS_CODES = {429, 500, 502, 503, 504}
|
||
|
||
def is_retryable_response(response: httpx.Response) -> bool:
|
||
return response.status_code in RETRY_STATUS_CODES
|
||
|
||
@retry(
|
||
retry=(
|
||
retry_if_exception_type(TRANSIENT_EXCEPTIONS) |
|
||
retry_if_result(is_retryable_response)
|
||
),
|
||
stop=stop_after_attempt(5),
|
||
wait=wait_exponential_jitter(initial=1, max=30),
|
||
before_sleep=before_sleep_log(logger, logging.WARNING),
|
||
)
|
||
def robust_http_call(
|
||
method: str,
|
||
url: str,
|
||
**kwargs,
|
||
) -> httpx.Response:
|
||
"""带全面重试处理的 HTTP 调用。"""
|
||
return httpx.request(method, url, timeout=30, **kwargs)
|
||
```
|
||
|
||
## 详细的完整示例与模式
|
||
|
||
详细章节（以 `## 高级模式` 开头）位于 `references/details.md` 中。当以上导航摘要不足以满足需求时，请阅读该文件。
|
||
|
||
## 最佳实践总结
|
||
|
||
1. **仅重试瞬时错误**——不要重试代码缺陷或认证失败
|
||
2. **使用指数退避**——给服务恢复的时间
|
||
3. **添加抖动**——防止同步重试导致惊群效应
|
||
4. **限制总时长**——`stop_after_attempt(5) | stop_after_delay(60)`
|
||
5. **记录每次重试**——静默重试会隐藏系统性问题
|
||
6. **使用装饰器**——将重试逻辑与业务逻辑分离
|
||
7. **注入依赖**——使基础设施可测试
|
||
8. **处处设置超时**——每个网络调用都需要超时
|
||
9. **优雅降级**——为非关键路径返回缓存/默认值
|
||
10. **监控重试率**——高重试率表明存在根本性问题
|