Python：装饰器、迭代器、生成器在真实项目里到底怎么用

发表于 分类于 本文字数： 2.9k 阅读时长 ≈ 10 分钟
装饰器、迭代器、生成器如果只按定义去学，很快就会停留在会写 demo、不会落地。这篇文章围绕一个实际的慢请求扫描脚本，把三者在日志处理、批量读取和执行耗时统计中的作用串起来。

装饰器、迭代器、生成器这三个词，一上来很容易显得 Python 开始变高级了。

真正的问题通常不是概念本身太难，而是学法容易跑偏：

  • 装饰器只会背 @xxx
  • 迭代器只记得 __iter__ 和 __next__
  • 生成器只会写 yield 1

这些都不能算真正会用。因为一进实际脚本，最关键的问题不是“能不能写出来”，而是：

  • 到底什么时候值得用装饰器
  • 什么时候应该一批一批读数据，而不是一次性全塞进内存
  • 什么时候更适合边产生结果边消费，而不是先构造一整个列表

这一篇直接围绕一个实际脚本展开：做一个慢请求扫描脚本。

这个脚本要完成这些事：

  1. 从访问日志里读取请求记录
  2. 一批一批扫描，避免一次性读太多
  3. 过滤出耗时超过阈值的请求
  4. 统计扫描耗时和结果数量

用这个场景把三块内容串起来：

  1. 装饰器用来给函数补耗时统计和统一日志
  2. 迭代器用来做批量读取
  3. 生成器用来边扫描边产出慢请求

一、这篇文章要解决什么问题

读完这一篇，应该能独立完成这些动作：

  • 给一个函数补统一的耗时统计
  • 用迭代器组织分批读取逻辑
  • 用生成器按需产生结果
  • 判断哪些地方应该返回列表，哪些地方更适合 yield
  • 排查几类典型错误，例如装饰器丢参数、生成器只能消费一次、迭代器不结束

如果这些动作能独立做出来，装饰器、迭代器、生成器就不再只是“看懂”，而是开始能落到实际脚本里。

二、先看要完成的实际功能

假设现在手里有这样一份访问日志：

1
2
3
4
5
6
7
records = [
    {"path": "/login", "cost_ms": 120, "status": 200},
    {"path": "/orders", "cost_ms": 980, "status": 200},
    {"path": "/health", "cost_ms": 20, "status": 200},
    {"path": "/pay", "cost_ms": 1500, "status": 500},
    {"path": "/orders", "cost_ms": 720, "status": 200},
]

现在要写一个脚本，扫描出慢请求，输出类似这样：

1
2
3
4
5
6
7
开始执行: scan_slow_requests
执行结束: scan_slow_requests, cost=0.0003s

慢请求数量: 3
/orders 980ms
/pay 1500ms
/orders 720ms

这个需求不复杂，但已经足够把三者真正接进实际功能。

三、先从最小版本开始

先别急着写装饰器和迭代器，先把最简单的扫描逻辑跑起来。

新建 slow_scan.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
records = [
    {"path": "/login", "cost_ms": 120, "status": 200},
    {"path": "/orders", "cost_ms": 980, "status": 200},
    {"path": "/health", "cost_ms": 20, "status": 200},
    {"path": "/pay", "cost_ms": 1500, "status": 500},
    {"path": "/orders", "cost_ms": 720, "status": 200},
]

slow_records = []

for record in records:
    if record["cost_ms"] >= 500:
        slow_records.append(record)

print(len(slow_records))
for item in slow_records:
    print(item["path"], item["cost_ms"])

执行：

1
python slow_scan.py

输出：

1
2
3
4
3
/orders 980
/pay 1500
/orders 720

这个版本当然能跑，但还没解决三个实际问题：

  • 没有统一记录执行耗时
  • 如果数据很多，无法分批处理
  • 所有结果先堆成列表，不能边产出边消费

四、装饰器最实际的用法是什么

第一次学装饰器时，注意力很容易全落在语法结构上：

1
2
3
@decorator
def func():
    ...

但装饰器最实际的价值，不是语法好看，而是：给一批函数补统一行为。

在这个脚本里，最适合统一补的行为就是执行耗时统计。

1. 先写一个最小装饰器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import time


def log_cost(func):
    def wrapper():
        start = time.time()
        print(f"开始执行: {func.__name__}")
        result = func()
        cost = time.time() - start
        print(f"执行结束: {func.__name__}, cost={cost:.4f}s")
        return result

    return wrapper


@log_cost
def run():
    time.sleep(0.1)
    return "ok"


print(run())

执行：

1
python slow_scan.py

输出：

1
2
3
开始执行: run
执行结束: run, cost=0.1001s
ok

这里装饰器的价值很清楚：

  • 业务函数自己不用管统计逻辑
  • 同样的统计逻辑可以复用到多个函数

2. 一个实际错误：忘了接收参数

错误写法：

1
2
3
4
5
6
7
8
9
10
11
12
def log_cost(func):
    def wrapper():
        return func()
    return wrapper


@log_cost
def add(a, b):
    return a + b


print(add(1, 2))

报错：

1
TypeError: wrapper() takes 0 positional arguments but 2 were given

原因很直接：

  • 被装饰的函数有参数
  • wrapper() 却没接参数

修复写法：

1
2
3
4
def log_cost(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

在实际项目里，只要装饰器要复用，*args, **kwargs 几乎就是默认写法。

3. 再补一个真实细节：保留函数名

如果不处理元数据：

1
print(run.__name__)

很可能输出：

1
wrapper

这会影响日志和调试。

实际写法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import time
from functools import wraps


def log_cost(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        print(f"开始执行: {func.__name__}")
        result = func(*args, **kwargs)
        cost = time.time() - start
        print(f"执行结束: {func.__name__}, cost={cost:.4f}s")
        return result

    return wrapper

五、迭代器什么时候值得自己写

如果只是普通遍历列表，直接 for 就够了，不需要自己写迭代器。

迭代器真正值得上场的场景通常是：

  • 想控制每次取多少数据
  • 想把“怎么取下一批”的逻辑封装起来
  • 想让调用方只关心遍历，不关心内部状态

在这个脚本里，最合适的例子就是分批读取日志。

1. 先写一个批量迭代器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class BatchReader:
    def __init__(self, items, batch_size):
        self.items = items
        self.batch_size = batch_size
        self.index = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.index >= len(self.items):
            raise StopIteration

        batch = self.items[self.index:self.index + self.batch_size]
        self.index += self.batch_size
        return batch

调用：

1
2
3
4
5
records = [1, 2, 3, 4, 5]
reader = BatchReader(records, batch_size=2)

for batch in reader:
    print(batch)

输出：

1
2
3
[1, 2]
[3, 4]
[5]

2. 一个实际错误：忘了结束条件

错误写法：

1
2
3
4
5
6
7
8
9
10
11
12
class BadReader:
    def __init__(self, items):
        self.items = items
        self.index = 0

    def __iter__(self):
        return self

    def __next__(self):
        batch = self.items[self.index:self.index + 2]
        self.index += 2
        return batch

这个类的问题是：切片超界时会返回空列表，但不会抛 StopIteration。

结果可能是：

  • for 循环一直跑
  • 或者逻辑消费到空批次后还要额外判断

修复关键点很简单：

1
2
if self.index >= len(self.items):
    raise StopIteration

3. 什么时候不该自己写迭代器

如果只是固定列表循环，直接：

1
2
for item in records:
    ...

就够了。

不要为了“用上迭代器”而硬写类。迭代器的价值在于封装遍历状态，不在于写法看起来高级。

六、生成器为什么适合做扫描脚本

生成器最实际的价值不是“语法短”，而是：结果可以边产生边消费，不需要先攒完整个列表。

在慢请求扫描场景里，这就很适合。

1. 先写一个最小生成器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def find_slow_records(records, threshold):
    for record in records:
        if record["cost_ms"] >= threshold:
            yield record


records = [
    {"path": "/login", "cost_ms": 120},
    {"path": "/orders", "cost_ms": 980},
    {"path": "/pay", "cost_ms": 1500},
]

for item in find_slow_records(records, 500):
    print(item)

输出：

1
2
{'path': '/orders', 'cost_ms': 980}
{'path': '/pay', 'cost_ms': 1500}

2. 为什么这里不直接返回列表

当然也可以写成：

1
2
3
4
5
6
def find_slow_records(records, threshold):
    result = []
    for record in records:
        if record["cost_ms"] >= threshold:
            result.append(record)
    return result

这在小数据时没问题。

但如果日志很多，生成器的优势就明显了：

  • 不需要先把全部结果存进内存
  • 消费端可以边拿边处理
  • 可以更自然地接在流式处理链路后面

3. 一个实际错误：把生成器当列表反复用

例如：

1
2
3
4
slow_iter = find_slow_records(records, 500)

print(list(slow_iter))
print(list(slow_iter))

输出通常是：

1
2
[{'path': '/orders', 'cost_ms': 980}, {'path': '/pay', 'cost_ms': 1500}]
[]

原因不是 Python 抽风，而是生成器本来就是一次性消费的迭代对象。

如果后面还要重复用，实际做法有两个：

  • 要么重新生成一次
  • 要么第一次就转成列表

例如：

1
2
3
slow_records = list(find_slow_records(records, 500))
print(slow_records)
print(slow_records)

七、把三者真正串起来

现在把装饰器、迭代器、生成器合到一个完整脚本里。

slow_scan.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
import time
from functools import wraps


def log_cost(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        print(f"开始执行: {func.__name__}")
        result = func(*args, **kwargs)
        cost = time.time() - start
        print(f"执行结束: {func.__name__}, cost={cost:.4f}s")
        return result

    return wrapper


class BatchReader:
    def __init__(self, items, batch_size):
        self.items = items
        self.batch_size = batch_size
        self.index = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.index >= len(self.items):
            raise StopIteration

        batch = self.items[self.index:self.index + self.batch_size]
        self.index += self.batch_size
        return batch


def find_slow_records(records, threshold):
    for record in records:
        if record["cost_ms"] >= threshold:
            yield record


@log_cost
def scan_slow_requests(records, threshold=500, batch_size=2):
    result = []
    reader = BatchReader(records, batch_size=batch_size)

    for batch in reader:
        for item in find_slow_records(batch, threshold):
            result.append(item)

    return result


def main():
    records = [
        {"path": "/login", "cost_ms": 120, "status": 200},
        {"path": "/orders", "cost_ms": 980, "status": 200},
        {"path": "/health", "cost_ms": 20, "status": 200},
        {"path": "/pay", "cost_ms": 1500, "status": 500},
        {"path": "/orders", "cost_ms": 720, "status": 200},
    ]

    slow_records = scan_slow_requests(records, threshold=500, batch_size=2)

    print(f"\n慢请求数量: {len(slow_records)}")
    for item in slow_records:
        print(item["path"], f'{item["cost_ms"]}ms')


if __name__ == "__main__":
    main()

执行：

1
python slow_scan.py

输出可能是：

1
2
3
4
5
6
7
开始执行: scan_slow_requests
执行结束: scan_slow_requests, cost=0.0000s

慢请求数量: 3
/orders 980ms
/pay 1500ms
/orders 720ms

八、怎么测试这一层是不是真的掌握了

这一层不能只看懂，要自己改一遍。

可以直接做这些动作：

  1. 把慢请求阈值从 500 改成 800
  2. 让装饰器额外打印结果数量
  3. 让批量读取器支持自定义起始位置
  4. 让生成器只产出状态码非 200 的慢请求
  5. 给扫描函数补两个最小测试

如果这些改动都能独立完成，说明三者已经开始真正会用了。

九、一个实际排错场景

这类脚本里一个非常常见的实际问题是：日志明明有慢请求，但最终结果却是空的。

例如第一版可能会这样写：

1
2
3
4
slow_iter = find_slow_records(records, 500)
print(list(slow_iter))
slow_records = list(slow_iter)
print(slow_records)

第二次打印结果是空的，于是把问题误判成过滤逻辑有问题。

这时排查顺序应该很直接：

  1. 先看 find_slow_records() 返回的是不是生成器
  2. 再看这个生成器是不是已经被消费过
  3. 如果后面还要复用，第一次就转成列表

如果把生成器误当成可以反复遍历的列表，结果就会看起来像“逻辑失效”，其实只是消费方式错了。

十、一个实际练习

可以直接把这一篇变成一个完整练习。

练习目标：做一个“错误日志扫描脚本”。

要求：

  1. 用装饰器记录扫描耗时
  2. 用迭代器按批读取日志
  3. 用生成器过滤出状态码为 500 的记录
  4. 统计每个接口的错误数量
  5. 输出扫描结果
  6. 补一个错误场景，例如生成器被重复消费

如果这个练习能独立做完，说明装饰器、迭代器、生成器已经开始真正进入实际脚本能力了。

十一、这篇文章学完以后，下一步应该补什么

如果这一篇已经能跟着做完，下一步最适合继续补的是：

  1. 异常处理、上下文管理和资源清理怎么写才稳
  2. 文件处理、序列化、时间处理和路径管理有哪些高频坑
  3. 项目从脚本到工程时，代码边界怎么继续拆

因为到这一步，已经开始进入“脚本可以工作，但怎样让它更稳、更省资源、更容易维护”的阶段。

十二、结语

装饰器、迭代器、生成器并不是三块孤立的“高级语法”。

在实际脚本里，它们经常是在解决同一类问题：

  • 装饰器负责补统一行为
  • 迭代器负责控制遍历方式
  • 生成器负责按需产出结果

只要把它们放进一个实际功能里一起练，就不会再停留在“看懂定义”，而会开始变成真正能用的脚本工具。