Python 生成器深度实战：用 yield 优雅处理百万级数据

1
def countdown(n):
2
    print(f"开始倒计时，从 {n} 开始")
3
    while n > 0:
4
        yield n  # 暂停，把 n 交出去
5
        n -= 1
6
        print(f"继续执行，n 现在是 {n}")
7
    print("倒计时结束！")
8

9
gen = countdown(3)
10
print(type(gen))  # <class 'generator'>
11

12
print(next(gen))  # 输出：开始倒计时，从 3 开始 → 返回 3
13
print(next(gen))  # 输出：继续执行，n 现在是 2 → 返回 2
14
print(next(gen))  # 输出：继续执行，n 现在是 1 → 返回 1
15
# next(gen)       # 输出：继续执行，n 现在是 0 → 倒计时结束！→ StopIteration

1
import sys
2
import tracemalloc
3

4
# 方式一：列表推导式
5
tracemalloc.start()
6
numbers_list = [x ** 2 for x in range(1_000_000)]
7
list_snapshot = tracemalloc.take_snapshot()
8
list_mem = sum(stat.size for stat in list_snapshot.statistics('filename'))
9
print(f"列表内存：{list_mem / 1024 / 1024:.1f} MB")
10
# 列表内存：约 40 MB
11
tracemalloc.stop()
12

13
# 方式二：生成器表达式
14
tracemalloc.start()
15
numbers_gen = (x ** 2 for x in range(1_000_000))
16
gen_snapshot = tracemalloc.take_snapshot()
17
gen_mem = sum(stat.size for stat in gen_snapshot.statistics('filename'))
18
print(f"生成器内存：{gen_mem / 1024:.1f} KB")
19
# 生成器内存：约 0.2 KB
20
tracemalloc.stop()

1
import csv
2
from datetime import datetime
3

4
def read_csv_rows(filepath):
5
    """第一步：读取 CSV"""
6
    with open(filepath) as f:
7
        reader = csv.DictReader(f)
8
        for row in reader:
9
            yield row
10

11
def filter_recent(rows, days=30):
12
    """第二步：过滤最近 N 天的记录"""
13
    cutoff = datetime.now().timestamp() - days * 86400
14
    for row in rows:
15
        if float(row['timestamp']) > cutoff:
16
            yield row
17

18
def transform_amount(rows):
19
    """第三步：转换金额字段"""
20
    for row in rows:
21
        row['amount'] = float(row['amount']) * 100  # 分转元
22
        row['amount_display'] = f"¥{row['amount']:.2f}"
23
        yield row
24

25
def aggregate_by_category(rows):
26
    """第四步：按类别汇总"""
27
    totals = {}
28
    for row in rows:
29
        cat = row['category']
30
        totals[cat] = totals.get(cat, 0) + row['amount']
31
    return totals  # 最后一步可以不是生成器
32

33
# 组装管道
34
pipeline = transform_amount(
35
    filter_recent(
36
        read_csv_rows("transactions.csv"),
37
        days=7
38
    )
39
)
40

41
# 数据流：文件 → 读取 → 过滤 → 转换 → 逐条处理
42
for record in pipeline:
43
    print(record['amount_display'])

1
from itertools import islice
2

3
def batched(iterable, batch_size):
4
    """将可迭代对象分成固定大小的批次
5

6
    Python 3.12+ 可以直接用 itertools.batched
7
    """
8
    iterator = iter(iterable)
9
    while True:
10
        batch = list(islice(iterator, batch_size))
11
        if not batch:
12
            break
13
        yield batch
14

15
# 使用示例：每 1000 条一批插入数据库
16
def bulk_insert(records):
17
    for batch in batched(records, 1000):
18
        db.execute_many(
19
            "INSERT INTO logs (time, level, msg) VALUES (?, ?, ?)",
20
            [(r['time'], r['level'], r['msg']) for r in batch]
21
        )
22
        print(f"插入了 {len(batch)} 条记录")
23

24
# records 可以是一个生成器，百万条数据也不怕
25
bulk_insert(read_errors("/var/log/huge.log"))

1
def running_average():
2
    """计算动态平均值"""
3
    total = 0
4
    count = 0
5
    average = None
6
    while True:
7
        value = yield average  # 接收外部发送的值，同时返回当前平均值
8
        if value is not None:
9
            total += value
10
            count += 1
11
            average = total / count
12

13
avg = running_average()
14
next(avg)          # 启动生成器（必须先调用一次 next）
15

16
print(avg.send(10))  # 10.0
17
print(avg.send(20))  # 15.0
18
print(avg.send(30))  # 20.0
19
print(avg.send(15))  # 18.75

1
class ReusableRange:
2
    """可重复遍历的自定义范围"""
3
    def __init__(self, start, stop):
4
        self.start = start
5
        self.stop = stop
6

7
    def __iter__(self):
8
        current = self.start
9
        while current < self.stop:
10
            yield current
11
            current += 1
12

13
nums = ReusableRange(0, 5)
14
print(list(nums))  # [0, 1, 2, 3, 4]
15
print(list(nums))  # [0, 1, 2, 3, 4] ← 可以重复遍历！

1
def gen_with_return():
2
    yield 1
3
    yield 2
4
    return "我是返回值"  # 大部分场景下这个值会被吞掉
5

6
g = gen_with_return()
7
print(next(g))  # 1
8
print(next(g))  # 2
9
try:
10
    next(g)
11
except StopIteration as e:
12
    print(e.value)  # "我是返回值" ← 藏在异常里
13

14
# 但 for 循环直接忽略
15
for x in gen_with_return():
16
    print(x)  # 只输出 1, 2

1
def managed_resource():
2
    """带资源管理的生成器"""
3
    print("获取资源...")
4
    resource = acquire_resource()
5
    try:
6
        while True:
7
            data = yield process(resource)
8
            if data == "STOP":
9
                break
10
    except GeneratorExit:
11
        print("生成器被关闭，清理资源...")
12
    finally:
13
        release_resource(resource)
14
        print("资源已释放")
15

16
gen = managed_resource()
17
next(gen)           # 启动，获取资源
18
gen.send("data1")   # 处理数据
19
gen.close()         # 触发 GeneratorExit → finally 块执行清理

1
from itertools import chain, compress, groupby, takewhile, dropwhile
2

3
# chain：串联多个可迭代对象
4
all_logs = chain(
5
    read_errors("app1.log"),
6
    read_errors("app2.log"),
7
    read_errors("app3.log")
8
)
9

10
# compress：根据选择器过滤
11
data = ['a', 'b', 'c', 'd', 'e']
12
selectors = [1, 0, 1, 0, 1]
13
print(list(compress(data, selectors)))  # ['a', 'c', 'e']
14

15
# groupby：分组（要求数据已排序）
16
records = sorted(records, key=lambda r: r['category'])
17
for category, group in groupby(records, key=lambda r: r['category']):
18
    items = list(group)
19
    print(f"{category}: {len(items)} 条")
20

21
# takewhile / dropwhile：条件截取
22
nums = [2, 4, 6, 7, 8, 10]
23
print(list(takewhile(lambda x: x % 2 == 0, nums)))  # [2, 4, 6]
24
print(list(dropwhile(lambda x: x % 2 == 0, nums)))  # [7, 8, 10]