文章

6 · 闭包与装饰器：函数的终极形态

#006 · 2026-04-16 · Python
PythonPython教程进阶函数装饰器闭包

🔗 知识图谱导航：阅读本文前，建议先掌握/回顾 《5 · 文本清洗：字符串高阶操作与正则》 中的核心概念；本文会在这个基础上继续推进。 承上启下：上一篇我们用正则清洗了文本。现在进入函数的深水区——装饰器是 Python 最优雅的特性之一，Flask/FastAPI 的路由、pytest 的 fixture 都基于它。

极客解析：先把数据流、控制流和模块边界跑通，再谈抽象；每段代码都围绕一个可执行 CLI 闭环展开。

痛点与架构

装饰器的本质：用一个函数包装另一个函数，不修改原函数代码，扩展其行为。

原函数 fn
    ↓ decorator(fn)
wrapper 函数（新功能 + 调用原函数）
    ↓ @decorator 语法糖
fn = decorator(fn)   ← 自动完成这一步
概念 一句话 典型用途
闭包 函数记住了它定义时的环境 计数器、缓存、工厂函数
装饰器 包装函数，扩展行为 计时、重试、权限、缓存
装饰器工厂 带参数的装饰器 @retry(max=3)

实战演练场

步步为营：核心逻辑自适应拆解

装饰器不是魔法，它就是“把一个函数交给另一个函数包装”。这一篇我们按“闭包记忆 -> 计时包装 -> 重试包装 -> 缓存包装 -> 业务函数组合 -> 性能报告”的顺序慢慢拆。

Step 1：用闭包做一个会记住上次调用时间的限速器

痛点与机制：

闭包最难的点是“函数为什么还能记住外面的变量”。你可以把 make_rate_limiter() 想成做了一个带秒表的保安，last_call_time 就是他手里的小纸条；每次有人进门，他先看看上次是什么时候放行的。

核心源码（逐字来自文末完整源码）：

def make_rate_limiter(calls_per_second: float) -> Callable:
    """
    闭包工厂：创建一个限速器。
    interval 变量被内部函数 limiter 捕获——这就是闭包。
    """
    interval = 1.0 / calls_per_second
    last_call_time: list[float] = [0.0]   # 用列表包装，允许 nonlocal 修改

    def limiter(fn: Callable) -> Callable:
        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            elapsed = time.time() - last_call_time[0]
            if elapsed < interval:
                time.sleep(interval - elapsed)
            last_call_time[0] = time.time()
            return fn(*args, **kwargs)
        return wrapper
    return limiter

可运行演示（补齐 Mock 数据与 print 反馈）：

import functools
import random
import time
from collections import defaultdict
from typing import Any, Callable, TypeVar

F = TypeVar("F", bound=Callable[..., Any])
_perf_records: dict[str, list[float]] = defaultdict(list)


def make_rate_limiter(calls_per_second: float) -> Callable:
    """
    闭包工厂：创建一个限速器。
    interval 变量被内部函数 limiter 捕获——这就是闭包。
    """
    interval = 1.0 / calls_per_second
    last_call_time: list[float] = [0.0]   # 用列表包装，允许 nonlocal 修改

    def limiter(fn: Callable) -> Callable:
        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            elapsed = time.time() - last_call_time[0]
            if elapsed < interval:
                time.sleep(interval - elapsed)
            last_call_time[0] = time.time()
            return fn(*args, **kwargs)
        return wrapper
    return limiter

@make_rate_limiter(5.0)
def fetch_task(task_id: int) -> str:
    return f"任务 {task_id} 已拉取"

start = time.perf_counter()
for i in range(3):
    print(fetch_task(i))
print(f"总耗时约: {time.perf_counter() - start:.2f} 秒")

Step 2：用 timer 给函数套一层秒表，不改原函数也能计时

痛点与机制：

@timer 的本质是 slow_add = timer(slow_add)。原函数像一杯咖啡，装饰器像给杯子套上隔热套：咖啡没变，但你拿它时多了“记录耗时”这个能力。

核心源码（逐字来自文末完整源码）：

def timer(fn: F) -> F:
    """记录函数每次调用的耗时，存入全局性能记录。"""
    @functools.wraps(fn)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        elapsed = time.perf_counter() - start
        _perf_records[fn.__name__].append(elapsed)
        return result
    return wrapper  # type: ignore[return-value]

可运行演示（补齐 Mock 数据与 print 反馈）：

import functools
import random
import time
from collections import defaultdict
from typing import Any, Callable, TypeVar

F = TypeVar("F", bound=Callable[..., Any])
_perf_records: dict[str, list[float]] = defaultdict(list)

# Step 2：装饰器像给函数套透明外壳，先计时，再调用原函数。
def timer(fn: F) -> F:
    """记录函数每次调用的耗时，存入全局性能记录。"""
    @functools.wraps(fn)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        elapsed = time.perf_counter() - start
        _perf_records[fn.__name__].append(elapsed)
        return result
    return wrapper  # type: ignore[return-value]

@timer
def slow_add(a: int, b: int) -> int:
    time.sleep(0.03)
    return a + b

print("计算结果:", slow_add(3, 5))
print("性能记录:", {k: [round(x, 4) for x in v] for k, v in _perf_records.items()})

Step 3：用 retry 装饰器工厂把“失败重试”参数化

痛点与机制：

retry(max_attempts=3, delay=0.01) 先把“最多试几次、隔多久再试”记下来，再生成真正包函数的装饰器。它像客服回拨机制：第一次没人接不会马上放弃，而是按规则再拨几次。

核心源码（逐字来自文末完整源码）：

def retry(max_attempts: int = 3, delay: float = 0.1, exceptions: tuple = (Exception,)):
    """
    带参数的装饰器（装饰器工厂）。
    用法：@retry(max_attempts=3, delay=0.5)
    """
    def decorator(fn: F) -> F:
        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            last_exc: Exception | None = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt < max_attempts:
                        print(f"  ⚠️  {fn.__name__} 第{attempt}次失败: {e}，{delay}s后重试")
                        time.sleep(delay)
            raise RuntimeError(f"{fn.__name__} 重试{max_attempts}次后仍失败") from last_exc
        return wrapper  # type: ignore[return-value]
    return decorator

可运行演示（补齐 Mock 数据与 print 反馈）：

import functools
import random
import time
from collections import defaultdict
from typing import Any, Callable, TypeVar

F = TypeVar("F", bound=Callable[..., Any])
_perf_records: dict[str, list[float]] = defaultdict(list)

# Step 3：装饰器工厂先接收配置，再返回真正的装饰器。
def retry(max_attempts: int = 3, delay: float = 0.1, exceptions: tuple = (Exception,)):
    """
    带参数的装饰器（装饰器工厂）。
    用法：@retry(max_attempts=3, delay=0.5)
    """
    def decorator(fn: F) -> F:
        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            last_exc: Exception | None = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt < max_attempts:
                        print(f"  ⚠️  {fn.__name__} 第{attempt}次失败: {e}，{delay}s后重试")
                        time.sleep(delay)
            raise RuntimeError(f"{fn.__name__} 重试{max_attempts}次后仍失败") from last_exc
        return wrapper  # type: ignore[return-value]
    return decorator

attempts = {"count": 0}

@retry(max_attempts=3, delay=0.01, exceptions=(ConnectionError,))
def flaky_service() -> str:
    attempts["count"] += 1
    if attempts["count"] < 3:
        raise ConnectionError("临时网络抖动")
    return "第三次终于成功"

print("调用结果:", flaky_service())
print("实际尝试次数:", attempts["count"])

Step 4：用 simple_cache 记住算过的结果，少做重复劳动

痛点与机制：

缓存的核心是“相同输入直接复用旧答案”。第一次算 square(4) 要真的计算，第二次再来 4 就像翻备忘录，速度更快。hits 是命中缓存，misses 是没找到只能重算。

核心源码（逐字来自文末完整源码）：

def simple_cache(maxsize: int = 128):
    """简化版 LRU 缓存装饰器，演示缓存原理。"""
    def decorator(fn: F) -> F:
        cache: dict[tuple, Any] = {}
        hits = misses = 0

        @functools.wraps(fn)
        def wrapper(*args: Any) -> Any:
            nonlocal hits, misses
            key = args
            if key in cache:
                hits += 1
                return cache[key]
            misses += 1
            result = fn(*args)
            if len(cache) >= maxsize:
                # 删除最旧的条目
                oldest_key = next(iter(cache))
                del cache[oldest_key]
            cache[key] = result
            return result

        def cache_info() -> str:
            return f"hits={hits}, misses={misses}, size={len(cache)}"

        wrapper.cache_info = cache_info  # type: ignore[attr-defined]
        return wrapper  # type: ignore[return-value]
    return decorator

可运行演示（补齐 Mock 数据与 print 反馈）：

import functools
import random
import time
from collections import defaultdict
from typing import Any, Callable, TypeVar

F = TypeVar("F", bound=Callable[..., Any])
_perf_records: dict[str, list[float]] = defaultdict(list)

# Step 4：缓存像备忘录，见过的入参直接查表，不再重新计算。
def simple_cache(maxsize: int = 128):
    """简化版 LRU 缓存装饰器，演示缓存原理。"""
    def decorator(fn: F) -> F:
        cache: dict[tuple, Any] = {}
        hits = misses = 0

        @functools.wraps(fn)
        def wrapper(*args: Any) -> Any:
            nonlocal hits, misses
            key = args
            if key in cache:
                hits += 1
                return cache[key]
            misses += 1
            result = fn(*args)
            if len(cache) >= maxsize:
                # 删除最旧的条目
                oldest_key = next(iter(cache))
                del cache[oldest_key]
            cache[key] = result
            return result

        def cache_info() -> str:
            return f"hits={hits}, misses={misses}, size={len(cache)}"

        wrapper.cache_info = cache_info  # type: ignore[attr-defined]
        return wrapper  # type: ignore[return-value]
    return decorator

@simple_cache(maxsize=2)
def square(n: int) -> int:
    print(f"真正计算: {n} * {n}")
    return n * n

for value in [4, 4, 5, 4]:
    print(f"square({value}) =", square(value))
print("缓存状态:", square.cache_info())

Step 5：把 timer 用到真实批处理函数上，自动收集耗时

痛点与机制：

这里开始进入“业务函数 + 装饰器”的组合。process_text_batch() 只关心处理文本批次，@timer 负责在外层悄悄记录耗时。新手可以先记住：装饰器负责横切能力，业务函数负责核心任务。

核心源码（逐字来自文末完整源码）：

@timer
def process_text_batch(batch_id: int, size: int = 100) -> dict[str, int]:
    """模拟文本批处理任务。"""
    time.sleep(random.uniform(0.01, 0.05))   # 模拟 IO 耗时
    return {"batch_id": batch_id, "processed": size, "errors": random.randint(0, 3)}

可运行演示（补齐 Mock 数据与 print 反馈）：

import functools
import random
import time
from collections import defaultdict
from typing import Any, Callable, TypeVar

F = TypeVar("F", bound=Callable[..., Any])
_perf_records: dict[str, list[float]] = defaultdict(list)

def timer(fn: F) -> F:
    """记录函数每次调用的耗时，存入全局性能记录。"""
    @functools.wraps(fn)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        elapsed = time.perf_counter() - start
        _perf_records[fn.__name__].append(elapsed)
        return result
    return wrapper  # type: ignore[return-value]

# Step 5：业务函数只写业务，耗时记录交给 @timer。
@timer
def process_text_batch(batch_id: int, size: int = 100) -> dict[str, int]:
    """模拟文本批处理任务。"""
    time.sleep(random.uniform(0.01, 0.05))   # 模拟 IO 耗时
    return {"batch_id": batch_id, "processed": size, "errors": random.randint(0, 3)}

random.seed(7)
for batch_id in range(3):
    print("批处理结果:", process_text_batch(batch_id, size=50))
print("已记录耗时次数:", len(_perf_records["process_text_batch"]))

Step 6：把 retry 和 timer 叠起来，处理不稳定接口

痛点与机制：

两个装饰器叠放时，执行关系像穿衣服：先穿里面的 @timer，再穿外面的 @retry。接口失败时，外层 retry 会接住异常并重试；只要某一次调用成功返回，内层 timer 就能把这次成功调用的耗时记录下来。

核心源码（逐字来自文末完整源码）：

@retry(max_attempts=3, delay=0.05, exceptions=(ConnectionError,))
@timer
def unstable_api_call(endpoint: str) -> str:
    """模拟不稳定的 API 调用（70% 概率失败）。"""
    if random.random() < 0.7:
        raise ConnectionError(f"连接 {endpoint} 超时")
    return f"✅ {endpoint} 响应成功"

可运行演示（补齐 Mock 数据与 print 反馈）：

import functools
import random
import time
from collections import defaultdict
from typing import Any, Callable, TypeVar

F = TypeVar("F", bound=Callable[..., Any])
_perf_records: dict[str, list[float]] = defaultdict(list)

def timer(fn: F) -> F:
    """记录函数每次调用的耗时，存入全局性能记录。"""
    @functools.wraps(fn)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        elapsed = time.perf_counter() - start
        _perf_records[fn.__name__].append(elapsed)
        return result
    return wrapper  # type: ignore[return-value]

def retry(max_attempts: int = 3, delay: float = 0.1, exceptions: tuple = (Exception,)):
    """
    带参数的装饰器（装饰器工厂）。
    用法：@retry(max_attempts=3, delay=0.5)
    """
    def decorator(fn: F) -> F:
        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            last_exc: Exception | None = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt < max_attempts:
                        print(f"  ⚠️  {fn.__name__} 第{attempt}次失败: {e}，{delay}s后重试")
                        time.sleep(delay)
            raise RuntimeError(f"{fn.__name__} 重试{max_attempts}次后仍失败") from last_exc
        return wrapper  # type: ignore[return-value]
    return decorator

# Step 6：多个装饰器像多层防护服，离函数最近的 @timer 先包，再由 @retry 包外层。
@retry(max_attempts=3, delay=0.05, exceptions=(ConnectionError,))
@timer
def unstable_api_call(endpoint: str) -> str:
    """模拟不稳定的 API 调用（70% 概率失败）。"""
    if random.random() < 0.7:
        raise ConnectionError(f"连接 {endpoint} 超时")
    return f"✅ {endpoint} 响应成功"

# seed=1 的随机序列会先失败、再成功，方便你看清“重试后成功”的全过程。
random.seed(1)
try:
    print("接口结果:", unstable_api_call("/api/v1/tasks"))
except RuntimeError as exc:
    print("接口最终失败:", exc)
print("性能记录函数名:", list(_perf_records.keys()))

Step 7：给耗时的向量计算加缓存，让重复文本秒回

痛点与机制：

机器学习里的向量计算通常很贵。这里用 time.sleep(0.02) 模拟慢操作，再用缓存避免重复算同一个文本。它像饭店常客点同一份套餐，厨房不用每次重新研究菜单，直接按熟单出餐。

核心源码（逐字来自文末完整源码）：

@simple_cache(maxsize=32)
@timer
def compute_embedding(text: str) -> list[float]:
    """模拟计算文本向量（耗时操作，适合缓存）。"""
    time.sleep(0.02)   # 模拟模型推理耗时
    return [hash(c) % 100 / 100 for c in text[:4]]

可运行演示（补齐 Mock 数据与 print 反馈）：

import functools
import random
import time
from collections import defaultdict
from typing import Any, Callable, TypeVar

F = TypeVar("F", bound=Callable[..., Any])
_perf_records: dict[str, list[float]] = defaultdict(list)

def timer(fn: F) -> F:
    """记录函数每次调用的耗时，存入全局性能记录。"""
    @functools.wraps(fn)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        elapsed = time.perf_counter() - start
        _perf_records[fn.__name__].append(elapsed)
        return result
    return wrapper  # type: ignore[return-value]

def simple_cache(maxsize: int = 128):
    """简化版 LRU 缓存装饰器，演示缓存原理。"""
    def decorator(fn: F) -> F:
        cache: dict[tuple, Any] = {}
        hits = misses = 0

        @functools.wraps(fn)
        def wrapper(*args: Any) -> Any:
            nonlocal hits, misses
            key = args
            if key in cache:
                hits += 1
                return cache[key]
            misses += 1
            result = fn(*args)
            if len(cache) >= maxsize:
                # 删除最旧的条目
                oldest_key = next(iter(cache))
                del cache[oldest_key]
            cache[key] = result
            return result

        def cache_info() -> str:
            return f"hits={hits}, misses={misses}, size={len(cache)}"

        wrapper.cache_info = cache_info  # type: ignore[attr-defined]
        return wrapper  # type: ignore[return-value]
    return decorator

# Step 7：向量计算假装是模型推理，同一句文本重复出现时最适合缓存。
@simple_cache(maxsize=32)
@timer
def compute_embedding(text: str) -> list[float]:
    """模拟计算文本向量（耗时操作，适合缓存）。"""
    time.sleep(0.02)   # 模拟模型推理耗时
    return [hash(c) % 100 / 100 for c in text[:4]]

for text in ["python", "python", "decorator", "python"]:
    print(f"{text} ->", compute_embedding(text))
print("缓存状态:", compute_embedding.cache_info())
print("计时记录次数:", len(_perf_records["compute_embedding"]))

Step 8：把所有性能记录汇总成终端报告

痛点与机制：

print_perf_report() 是最后的仪表盘。前面的装饰器一直在后台记账，这里把调用次数、总耗时、平均耗时和峰值一次性列出来。对新手来说，这一步能把“装饰器做了什么”变成看得见的表格。

核心源码（逐字来自文末完整源码）：

def print_perf_report() -> None:
    if not _perf_records:
        print("  暂无性能数据")
        return
    print(f"\n  {'函数名':<25} {'调用':<6} {'总耗时':<10} {'均值':<10} {'峰值'}")
    print(f"  {'─'*25} {'─'*6} {'─'*10} {'─'*10} {'─'*10}")
    for name, times in sorted(_perf_records.items()):
        total = sum(times)
        avg = total / len(times)
        peak = max(times)
        print(f"  {name:<25} {len(times):<6} {total:<10.4f} {avg:<10.4f} {peak:.4f}")

可运行演示（补齐 Mock 数据与 print 反馈）：

import functools
import random
import time
from collections import defaultdict
from typing import Any, Callable, TypeVar

F = TypeVar("F", bound=Callable[..., Any])
_perf_records: dict[str, list[float]] = defaultdict(list)

# Step 8：报告函数只读 _perf_records，把零散耗时整理成表格。
def print_perf_report() -> None:
    if not _perf_records:
        print("  暂无性能数据")
        return
    print(f"\n  {'函数名':<25} {'调用':<6} {'总耗时':<10} {'均值':<10} {'峰值'}")
    print(f"  {'─'*25} {'─'*6} {'─'*10} {'─'*10} {'─'*10}")
    for name, times in sorted(_perf_records.items()):
        total = sum(times)
        avg = total / len(times)
        peak = max(times)
        print(f"  {name:<25} {len(times):<6} {total:<10.4f} {avg:<10.4f} {peak:.4f}")

_perf_records["load_data"].extend([0.02, 0.03, 0.01])
_perf_records["train_model"].extend([0.12, 0.15])
print_perf_report()

极客实战：完整源码与运行

现在，把上面的积木拼起来，将以下完整代码放进你的编辑器，运行它。先看整体闭环，再回头逐段改参数，你会更容易建立工程直觉。

# fn_probe.py
"""
函数性能探针工具链 —— 演示闭包、装饰器、装饰器工厂。
用法：
    python3 fn_probe.py
    python3 fn_probe.py --mode timer
    python3 fn_probe.py --mode retry
    python3 fn_probe.py --mode cache
"""

import argparse
import functools
import time
import random
from collections import defaultdict
from typing import Any, Callable, TypeVar

F = TypeVar("F", bound=Callable[..., Any])

# ── 全局性能记录（闭包共享状态）─────────────────────────────
_perf_records: dict[str, list[float]] = defaultdict(list)


# ── 1. 闭包：函数工厂 ─────────────────────────────────────────
def make_rate_limiter(calls_per_second: float) -> Callable:
    """
    闭包工厂：创建一个限速器。
    interval 变量被内部函数 limiter 捕获——这就是闭包。
    """
    interval = 1.0 / calls_per_second
    last_call_time: list[float] = [0.0]   # 用列表包装，允许 nonlocal 修改

    def limiter(fn: Callable) -> Callable:
        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            elapsed = time.time() - last_call_time[0]
            if elapsed < interval:
                time.sleep(interval - elapsed)
            last_call_time[0] = time.time()
            return fn(*args, **kwargs)
        return wrapper
    return limiter


# ── 2. 基础装饰器：计时器 ─────────────────────────────────────
def timer(fn: F) -> F:
    """记录函数每次调用的耗时，存入全局性能记录。"""
    @functools.wraps(fn)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        elapsed = time.perf_counter() - start
        _perf_records[fn.__name__].append(elapsed)
        return result
    return wrapper  # type: ignore[return-value]


# ── 3. 装饰器工厂：重试 ───────────────────────────────────────
def retry(max_attempts: int = 3, delay: float = 0.1, exceptions: tuple = (Exception,)):
    """
    带参数的装饰器（装饰器工厂）。
    用法：@retry(max_attempts=3, delay=0.5)
    """
    def decorator(fn: F) -> F:
        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            last_exc: Exception | None = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt < max_attempts:
                        print(f"  ⚠️  {fn.__name__} 第{attempt}次失败: {e}，{delay}s后重试")
                        time.sleep(delay)
            raise RuntimeError(f"{fn.__name__} 重试{max_attempts}次后仍失败") from last_exc
        return wrapper  # type: ignore[return-value]
    return decorator


# ── 4. 装饰器工厂：LRU 缓存（手动实现原理）──────────────────
def simple_cache(maxsize: int = 128):
    """简化版 LRU 缓存装饰器，演示缓存原理。"""
    def decorator(fn: F) -> F:
        cache: dict[tuple, Any] = {}
        hits = misses = 0

        @functools.wraps(fn)
        def wrapper(*args: Any) -> Any:
            nonlocal hits, misses
            key = args
            if key in cache:
                hits += 1
                return cache[key]
            misses += 1
            result = fn(*args)
            if len(cache) >= maxsize:
                # 删除最旧的条目
                oldest_key = next(iter(cache))
                del cache[oldest_key]
            cache[key] = result
            return result

        def cache_info() -> str:
            return f"hits={hits}, misses={misses}, size={len(cache)}"

        wrapper.cache_info = cache_info  # type: ignore[attr-defined]
        return wrapper  # type: ignore[return-value]
    return decorator


# ── 被装饰的业务函数 ──────────────────────────────────────────
@timer
def process_text_batch(batch_id: int, size: int = 100) -> dict[str, int]:
    """模拟文本批处理任务。"""
    time.sleep(random.uniform(0.01, 0.05))   # 模拟 IO 耗时
    return {"batch_id": batch_id, "processed": size, "errors": random.randint(0, 3)}


@retry(max_attempts=3, delay=0.05, exceptions=(ConnectionError,))
@timer
def unstable_api_call(endpoint: str) -> str:
    """模拟不稳定的 API 调用（70% 概率失败）。"""
    if random.random() < 0.7:
        raise ConnectionError(f"连接 {endpoint} 超时")
    return f"✅ {endpoint} 响应成功"


@simple_cache(maxsize=32)
@timer
def compute_embedding(text: str) -> list[float]:
    """模拟计算文本向量（耗时操作，适合缓存）。"""
    time.sleep(0.02)   # 模拟模型推理耗时
    return [hash(c) % 100 / 100 for c in text[:4]]


# ── 性能报告 ──────────────────────────────────────────────────
def print_perf_report() -> None:
    if not _perf_records:
        print("  暂无性能数据")
        return
    print(f"\n  {'函数名':<25} {'调用':<6} {'总耗时':<10} {'均值':<10} {'峰值'}")
    print(f"  {'─'*25} {'─'*6} {'─'*10} {'─'*10} {'─'*10}")
    for name, times in sorted(_perf_records.items()):
        total = sum(times)
        avg = total / len(times)
        peak = max(times)
        print(f"  {name:<25} {len(times):<6} {total:<10.4f} {avg:<10.4f} {peak:.4f}")


def demo_timer() -> None:
    print("\n  ── 计时装饰器演示 ────────────────────────")
    for i in range(5):
        result = process_text_batch(i, size=200)
        print(f"  Batch {i}: {result}")
    print_perf_report()


def demo_retry() -> None:
    print("\n  ── 重试装饰器演示 ────────────────────────")
    random.seed(42)
    try:
        result = unstable_api_call("/api/v1/tasks")
        print(f"  最终结果: {result}")
    except RuntimeError as e:
        print(f"  最终失败: {e}")


def demo_cache() -> None:
    print("\n  ── 缓存装饰器演示 ────────────────────────")
    texts = ["python", "machine learning", "python", "deep learning", "python"]
    for text in texts:
        vec = compute_embedding(text)
        print(f"  '{text}' → {vec[:3]}...")
    print(f"\n  缓存状态: {compute_embedding.cache_info()}")  # type: ignore[attr-defined]


def main() -> None:
    parser = argparse.ArgumentParser(description="函数性能探针工具链")
    parser.add_argument("--mode", choices=["timer", "retry", "cache", "all"], default="all")
    args = parser.parse_args()

    if args.mode in ("timer", "all"):
        demo_timer()
    if args.mode in ("retry", "all"):
        demo_retry()
    if args.mode in ("cache", "all"):
        demo_cache()


if __name__ == "__main__":
    main()

终端预期输出（--mode cache）：

$ python3 fn_probe.py --mode cache

  ── 缓存装饰器演示 ────────────────────────
  'python' → [0.12, 0.34, 0.56]...
  'machine learning' → [0.45, 0.23, 0.67]...
  'python' → [0.12, 0.34, 0.56]...   ← 命中缓存，无延迟
  'deep learning' → [0.78, 0.12, 0.34]...
  'python' → [0.12, 0.34, 0.56]...   ← 命中缓存

  缓存状态: hits=2, misses=3, size=3

装饰器执行顺序

# 装饰器叠加顺序演示
def decorator_a(fn):
    def wrapper(*a, **kw):
        print("A before"); result = fn(*a, **kw); print("A after"); return result
    return wrapper

def decorator_b(fn):
    def wrapper(*a, **kw):
        print("B before"); result = fn(*a, **kw); print("B after"); return result
    return wrapper

@decorator_a
@decorator_b
def fn(): print("fn()")

# 等价于：fn = decorator_a(decorator_b(fn))
fn()

NexDo Time ⚡

5 分钟极客微操：给 timer 装饰器增加 threshold 参数（变成装饰器工厂），当某次调用耗时超过阈值时，自动打印警告：⚠️ process_text_batch 耗时 0.08s，超过阈值 0.05s。

Don’t wait for next time, do it in the next moment.