从零实现一个可观测的 Python 异步任务重试器：超时、退避、日志与指标

声明：本文由 AI 辅助生成并经过人工可读性整理，示例代码仅用于技术学习与工程实践参考。

在业务系统里，重试几乎无处不在：调用第三方接口、发送消息、拉取文件、执行定时任务、访问数据库临时失败时，我们都希望系统“再试一次”。

但重试不是简单地 for 循环几次。一个生产可用的重试器至少要回答 4 个问题：

  1. 失败后什么时候再试？
  2. 每次调用最长允许执行多久？
  3. 重试过程如何记录日志？
  4. 出问题后能不能看到指标？

这篇文章从零实现一个 Python 异步任务重试器，覆盖：

  • asyncio 超时控制
  • 指数退避 Exponential Backoff
  • 随机抖动 Jitter
  • 结构化日志
  • 简单指标统计
  • 装饰器封装

最终效果如下：

@retry_async(max_attempts=4,timeout=2,base_delay=0.3,max_delay=3,retry_on=(TimeoutError, ConnectionError),
)
async def call_remote_service(order_id: str):...

调用方只关心业务函数，重试策略由装饰器统一处理。

一、为什么不能直接 while 重试？

最常见的写法可能是这样：

for i in range(3):try:return await do_work()except Exception:await asyncio.sleep(1)

这段代码能跑，但问题很多：

  • 捕获了所有异常，可能把参数错误、权限错误也重试了；
  • 固定 sleep，多个任务同时失败时会同时再次冲击下游；
  • 没有单次超时，任务卡住后根本不会进入下一轮；
  • 没有日志上下文，排查问题只能猜；
  • 没有指标，无法判断失败率是否升高。

所以我们需要把“重试”从业务代码里抽出来，做成一个可复用的小组件。

二、定义重试配置

先定义一个配置对象，把重试策略收敛起来。

from dataclasses import dataclass
from typing import tuple, Type@dataclass(frozen=True)
class RetryConfig:max_attempts: int = 3timeout: float = 3.0base_delay: float = 0.2max_delay: float = 5.0jitter: float = 0.2retry_on: tuple[Type[BaseException], ...] = (Exception,)

字段含义：

字段 说明
max_attempts 最大尝试次数，包含第一次执行
timeout 单次调用超时时间
base_delay 第一次重试前等待时间
max_delay 最大等待时间，避免指数退避无限变大
jitter 随机抖动比例，避免集中重试
retry_on 哪些异常允许重试

这里要注意：max_attempts=3 不是“失败后再重试 3 次”，而是“一共最多执行 3 次”。这个定义更容易和日志、指标对齐。

三、实现指数退避和随机抖动

指数退避的基本公式：

delay = base_delay * 2 ^ (attempt - 1)

例如 base_delay=0.2，重试等待时间大概是：

0.2s, 0.4s, 0.8s, 1.6s ...

再加一个最大值保护：

delay = min(delay, max_delay)

如果所有失败任务都按同一个节奏重试，会造成“羊群效应”。因此我们给 delay 加一点随机抖动：

import randomdef calculate_delay(attempt: int, base_delay: float, max_delay: float, jitter: float) -> float:delay = min(base_delay * (2 ** (attempt - 1)), max_delay)if jitter <= 0:return delayoffset = delay * jitterreturn random.uniform(delay - offset, delay + offset)

当 jitter=0.2 时，原本 1 秒的等待时间会随机落在 0.8 到 1.2 秒之间。

四、加一个最小指标收集器

生产环境一般会接 Prometheus、StatsD 或 OpenTelemetry。为了示例清晰，先实现一个内存版指标收集器。

from collections import defaultdict
from dataclasses import dataclass, field@dataclass
class RetryMetrics:total_calls: int = 0success_calls: int = 0failed_calls: int = 0retry_count: int = 0timeout_count: int = 0exception_count: dict[str, int] = field(default_factory=lambda: defaultdict(int))def snapshot(self) -> dict:return {"total_calls": self.total_calls,"success_calls": self.success_calls,"failed_calls": self.failed_calls,"retry_count": self.retry_count,"timeout_count": self.timeout_count,"exception_count": dict(self.exception_count),}

这个版本不追求复杂，只记录 6 类信息：

  • 总调用次数
  • 成功次数
  • 最终失败次数
  • 发生过多少次重试
  • 超时次数
  • 各类异常出现次数

后续如果要接 Prometheus，只需要在这些事件点上换成 Counter、Histogram 即可。

五、实现核心 retry 函数

现在开始写核心逻辑。

import asyncio
import functools
import logging
import time
from typing import Awaitable, Callable, TypeVar, ParamSpecP = ParamSpec("P")
R = TypeVar("R")logger = logging.getLogger("retry")
metrics = RetryMetrics()async def run_with_retry(func: Callable[P, Awaitable[R]],config: RetryConfig,*args: P.args,**kwargs: P.kwargs,
) -> R:metrics.total_calls += 1last_error: BaseException | None = Nonefor attempt in range(1, config.max_attempts + 1):started = time.perf_counter()try:result = await asyncio.wait_for(func(*args, **kwargs),timeout=config.timeout,)cost_ms = round((time.perf_counter() - started) * 1000, 2)metrics.success_calls += 1logger.info("task success",extra={"func": func.__name__,"attempt": attempt,"cost_ms": cost_ms,},)return resultexcept asyncio.TimeoutError as exc:last_error = TimeoutError(f"task timeout after {config.timeout}s")metrics.timeout_count += 1metrics.exception_count["TimeoutError"] += 1except config.retry_on as exc:last_error = excmetrics.exception_count[type(exc).__name__] += 1except Exception:metrics.failed_calls += 1logger.exception("task failed with non-retryable exception",extra={"func": func.__name__, "attempt": attempt},)raisecost_ms = round((time.perf_counter() - started) * 1000, 2)if attempt >= config.max_attempts:metrics.failed_calls += 1logger.error("task exhausted retry attempts",extra={"func": func.__name__,"attempt": attempt,"max_attempts": config.max_attempts,"cost_ms": cost_ms,"error": repr(last_error),},)raise last_errordelay = calculate_delay(attempt=attempt,base_delay=config.base_delay,max_delay=config.max_delay,jitter=config.jitter,)metrics.retry_count += 1logger.warning("task retry scheduled",extra={"func": func.__name__,"attempt": attempt,"next_attempt": attempt + 1,"delay": round(delay, 3),"cost_ms": cost_ms,"error": repr(last_error),},)await asyncio.sleep(delay)raise RuntimeError("unreachable retry state")

这里有几个关键点。

1. 用 asyncio.wait_for 控制单次超时

result = await asyncio.wait_for(func(*args, **kwargs), timeout=config.timeout)

这样每次尝试都有独立超时，不会因为某一次调用卡死导致整个任务挂住。

2. 只重试指定异常

except config.retry_on as exc:

不是所有异常都应该重试。

适合重试的异常：

  • 网络连接临时失败
  • 请求超时
  • 下游限流
  • 资源暂时不可用

不适合重试的异常：

  • 参数错误
  • 鉴权失败
  • 数据格式错误
  • 业务规则不允许

3. 最后一次失败必须抛出

重试器不能吞异常。否则调用方会误以为任务成功了。

raise last_error

六、封装成装饰器

为了让业务代码更干净，我们再包一层装饰器。

def retry_async(max_attempts: int = 3,timeout: float = 3.0,base_delay: float = 0.2,max_delay: float = 5.0,jitter: float = 0.2,retry_on: tuple[type[BaseException], ...] = (Exception,),
):config = RetryConfig(max_attempts=max_attempts,timeout=timeout,base_delay=base_delay,max_delay=max_delay,jitter=jitter,retry_on=retry_on,)def decorator(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:@functools.wraps(func)async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:return await run_with_retry(func, config, *args, **kwargs)return wrapperreturn decorator

业务代码可以直接这样使用：

@retry_async(max_attempts=4, timeout=1.5, retry_on=(ConnectionError, TimeoutError))
async def fetch_user(user_id: int) -> dict:return await request_user_from_remote(user_id)

七、完整可运行 Demo

下面给一个完整示例：前两次模拟失败，第三次成功。

import asyncio
import logginglogging.basicConfig(level=logging.INFO,format="%(asctime)s %(levelname)s %(name)s %(message)s",
)counter = {"value": 0}@retry_async(max_attempts=5,timeout=1,base_delay=0.2,max_delay=2,jitter=0.3,retry_on=(ConnectionError, TimeoutError),
)
async def unstable_api() -> str:counter["value"] += 1if counter["value"] <= 2:raise ConnectionError("remote service temporarily unavailable")await asyncio.sleep(0.1)return "ok"async def main():result = await unstable_api()print("result:", result)print("metrics:", metrics.snapshot())if __name__ == "__main__":asyncio.run(main())

可能输出：

WARNING retry task retry scheduled
WARNING retry task retry scheduled
INFO retry task success
result: ok
metrics: {'total_calls': 1,'success_calls': 1,'failed_calls': 0,'retry_count': 2,'timeout_count': 0,'exception_count': {'ConnectionError': 2}
}

八、生产环境还要补什么？

上面的实现已经够日常项目使用，但如果要放进核心链路，还建议继续补 5 个能力。

1. 总超时

目前控制的是“单次超时”，不是“整体超时”。

如果 max_attempts=5，每次 timeout=3s，中间还有退避等待，整体耗时可能超过十几秒。

核心接口建议再加一个 total_timeout。

2. 幂等保护

重试的前提是任务可以安全重复执行。

例如：

  • 查询接口：通常可以重试；
  • 创建订单：必须有幂等键；
  • 扣款操作：不能简单重试；
  • 发短信：要避免重复发送。

重试器只能解决技术失败，不能替你解决业务幂等。

3. 限流与熔断

如果下游已经故障，大量重试会放大压力。

更完整的方案应该是：

超时控制 + 重试 + 限流 + 熔断 + 降级

重试不是越多越好。很多场景下，快速失败比盲目等待更可靠。

4. 指标接入 Prometheus

建议至少打这些指标：

  • task_retry_total
  • task_success_total
  • task_failed_total
  • task_timeout_total
  • task_duration_seconds
  • task_retry_delay_seconds

有了指标后，才能设置告警：

  • 失败率突然升高；
  • 重试次数暴涨；
  • 任务耗时长尾变高；
  • 超时数量异常。

5. TraceId 贯穿日志

线上排查时，只有函数名和异常还不够。最好把请求 ID、订单 ID、用户 ID 等上下文写进日志。

可以用 contextvars 存 TraceId：

from contextvars import ContextVartrace_id_var = ContextVar("trace_id", default="-")

然后在日志里带上：

"trace_id": trace_id_var.get()

这样才能把一次请求里的所有重试日志串起来。

九、我的取舍建议

如果你只是写一个内部脚本：

最大尝试 3 次 + 固定超时 + 简单日志

就够了。

如果你在写后端服务：

指定异常重试 + 指数退避 + 抖动 + 指标

是最低配置。

如果你在写核心交易链路：

先做幂等，再谈重试；先有限流熔断，再放开重试次数。

重试的目标不是“最终一定成功”，而是“在短暂故障时提高成功率，同时不把故障扩大”。

十、总结

本文从零实现了一个 Python 异步任务重试器，核心能力包括：

  • 用 asyncio.wait_for 控制单次超时；
  • 用指数退避降低下游压力；
  • 用随机抖动避免集中重试；
  • 用日志记录每次尝试的上下文；
  • 用指标统计成功、失败、超时和重试次数；
  • 用装饰器降低业务接入成本。

真正可用的重试器，重点不在“多试几次”，而在“只对合适的失败重试，并让重试过程可观测”。

把这套思路抽成公共组件后，业务代码会更干净，线上排障也会轻松很多。