从零开始系统学习 Python 编程

Appearance

03-生成器基础 ​

Python 版本要求：Python 3.11+

贯穿项目：日志分析系统
本节目标：用 yield 实现日志解析生成器，内存友好

概念铺垫 ​

为什么需要生成器？ ​

问题场景：处理千万级数据 ​

python
import sys
from typing import Iterator

# ❌ 列表方式：内存爆炸
def squares_list(n: int) -> list[int]:
    return [x ** 2 for x in range(n)]

result = squares_list(10_000_000)
print(f"列表内存：{sys.getsizeof(result) / 1024 / 1024:.2f} MB")  # 约 80 MB！

# ✅ 生成器方式：内存友好
def squares_generator(n: int) -> Iterator[int]:
    for x in range(n):
        yield x ** 2

gen = squares_generator(10_000_000)
print(f"生成器内存：{sys.getsizeof(gen)} 字节")  # 约 200 字节！

生成器的优势：

  • 惰性求值：只在需要时才计算
  • 内存友好：不一次性生成所有数据
  • 代码简洁：用 yield 代替复杂的迭代器类

yield 概念：暂停点打卡 ​

生活类比 ​

┌─────────────────────────────────────────┐
│  yield vs return                         │
├─────────────────────────────────────────┤
│                                         │
│  return = 下班打卡                       │
│  • 做完就走                              │
│  • 不再回来                              │
│  • 只返回一次                            │
│                                         │
│  yield = 暂停点打卡                      │
│  • 做一步，歇一步                        │
│  • 下次从这里继续                        │
│  • 可以返回多次                          │
│                                         │
│  示例：                                   │
│  def work():                             │
│      yield "做完第1步"  # 打卡，暂停     │
│      yield "做完第2步"  # 继续，再打卡   │
│      yield "做完第3步"  # 继续，再打卡   │
│                                         │
└─────────────────────────────────────────┘

一句话：

yield = 暂停并返回值，下次从这里继续执行

yield vs return ​

python
from typing import Iterator

def with_return(n: int) -> int:
    """return：立即结束"""
    for i in range(n):
        return i  # 只返回第一个值

def with_yield(n: int) -> Iterator[int]:
    """yield：暂停返回"""
    for i in range(n):
        yield i  # 返回值，暂停等待下次

print(with_return(5))       # 0（只返回一次）
print(list(with_yield(5)))  # [0, 1, 2, 3, 4]（返回多次）

L1 理解层：会用 ​

yield 执行流程时间线 ​

┌─────────────────────────────────────────────────────────────┐
│  yield 执行时间线                                             │
│                                                             │
│  代码：                                      │
│      yield 1                                                │
│      yield 2                                                │
│      yield 3                                                │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │                                                      │   │
│  │  gen = simple_gen() ─→ 创建生成器                    │   │
│  │        ⚠️ 函数不执行！只是创建对象                     │   │
│  │        │                                             │   │
│  │        ↓                                             │   │
│  │  next(gen) ────────→ 执行到 yield 1                 │   │
│  │        │               返回 1，暂停                  │   │
│  │        │               状态：GEN_SUSPENDED           │   │
│  │        ↓                                             │   │
│  │  next(gen) ────────→ 从暂停处继续                   │   │
│  │        │               执行到 yield 2                │   │
│  │        │               返回 2，暂停                  │   │
│  │        ↓                                             │   │
│  │  next(gen) ────────→ 从暂停处继续                   │   │
│  │        │               执行到 yield 3                │   │
│  │        │               返回 3，暂停                  │   │
│  │        ↓                                             │   │
│  │  next(gen) ────────→ 从暂停处继续                   │   │
│  │        │               函数结束                      │   │
│  │        │               抛出 StopIteration            │   │
│  │        │               状态：GEN_CLOSED              │   │
│  │                                                      │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ⚠️ 关键：创建生成器时函数不执行                             │
│  ⚠️ 关键：首次 next() 才开始执行                             │
│                                                             │
└─────────────────────────────────────────────────────────────┘

最简示例 ​

python
from typing import Iterator

def simple_generator() -> Iterator[int]:
    yield 1
    yield 2
    yield 3

gen = simple_generator()  # 创建生成器（函数不执行！）
print(next(gen))  # 1
print(next(gen))  # 2
print(next(gen))  # 3
print(next(gen, "耗尽"))  # "耗尽"

带日志的执行流程 ​

python
from typing import Iterator

def logged_generator() -> Iterator[int]:
    print(">>> 开始执行")
    yield 1
    print(">>> 继续执行")
    yield 2
    print(">>> 再次执行")
    yield 3
    print(">>> 结束执行")

gen = logged_generator()
print("生成器已创建")      # 函数不执行！

print(f"返回: {next(gen)}")  # >>> 开始执行，返回: 1
print(f"返回: {next(gen)}")  # >>> 继续执行，返回: 2
print(f"返回: {next(gen)}")  # >>> 再次执行，返回: 3
print(f"返回: {next(gen, '耗尽')}")  # >>> 结束执行，返回: 耗尽

贯穿实战：日志解析生成器 ​

python
from typing import Iterator
from datetime import datetime

def parse_log_entries(filename: str) -> Iterator[dict]:
    """解析日志文件的生成器
    
    贯穿项目：日志分析系统第3步
    功能：逐行解析日志，产出结构化数据
    内存：无论文件多大，只占用当前行的内存
    
    日志格式示例：
    [2024-01-19 10:00:00] ERROR user_login - 用户登录失败
    """
    with open(filename, "r", encoding="utf-8") as f:
        for line in f:
            if not line.strip():
                continue
            
            if line.startswith("["):
                timestamp_str = line[1:20]
                try:
                    timestamp = datetime.strptime(
                        timestamp_str, "%Y-%m-%d %H:%M:%S"
                    )
                except ValueError:
                    timestamp = None
                
                parts = line[22:].split(" - ", 1)
                level = parts[0].strip() if parts else "UNKNOWN"
                message = parts[1] if len(parts) > 1 else ""
                
                yield {
                    "timestamp": timestamp,
                    "level": level,
                    "message": message.strip(),
                    "raw": line.strip()
                }

# 使用：处理大日志文件
for entry in parse_log_entries("app.log"):
    if entry["level"] == "ERROR":
        print(f"{entry['timestamp']}: {entry['message'][:50]}")

关键代码说明：

代码含义
for line in f文件对象是迭代器，逐行读取
yield {...}每解析一行就产出，不等待全部解析完
try/except ValueError跳过格式错误的时间戳，继续处理

L1 理解层：生成器基础 ​

生成器函数示例 ​

倒数生成器 ​

python
from typing import Iterator

def countdown(n: int) -> Iterator[int]:
    """倒数生成器"""
    while n > 0:
        yield n
        n -= 1

for num in countdown(5):
    print(num, end=" ")
# 5 4 3 2 1

斐波那契生成器 ​

python
from typing import Iterator

def fibonacci(limit: int | None = None) -> Iterator[int]:
    """斐波那契数列生成器"""
    a, b = 0, 1
    count = 0
    while limit is None or count < limit:
        yield a
        a, b = b, a + b
        count += 1

for num in fibonacci(10):
    print(num, end=" ")
# 0 1 1 2 3 5 8 13 21 34

无限素数生成器 ​

python
from typing import Iterator
import math

def primes() -> Iterator[int]:
    """无限素数生成器"""
    n = 2
    while True:
        if all(n % i != 0 for i in range(2, int(math.sqrt(n)) + 1)):
            yield n
        n += 1

for i, prime in enumerate(primes()):
    if i >= 10:
        break
    print(prime, end=" ")
# 2 3 5 7 11 13 17 19 23 29

生成器表达式 ​

python
from typing import Iterator

# 列表推导式：立即生成（方括号）
squares_list = [x ** 2 for x in range(5)]
print(squares_list)       # [0, 1, 4, 9, 16]

# 生成器表达式：惰性生成（圆括号）
squares_gen = (x ** 2 for x in range(5))
print(list(squares_gen))  # [0, 1, 4, 9, 16]

内存对比：

python
import sys

list_data = [x for x in range(1_000_000)]
gen_data = (x for x in range(1_000_000))

print(f"列表：{sys.getsizeof(list_data) / 1024:.2f} KB")  # 约 8 MB
print(f"生成器：{sys.getsizeof(gen_data)} 字节")         # 约 200 字节

L2 实践层：用好 ​

推荐做法 ​

做法原因示例
生成器函数代替迭代器类代码简洁，Python 自动处理协议def gen(): yield x
生成器表达式用于惰性计算内存友好(x ** 2 for x in range(n))
列表推导式用于多次访问列表可重复遍历[x ** 2 for x in range(n)]
用 yield from 委托子生成器简化嵌套迭代yield from sub_gen()
用 finally 清理资源生成器关闭时执行try: yield x; finally: cleanup()

反模式：不要这样做 ​

python
# ❌ 生成器表达式只为了遍历（不创建有意义结果）
result = (print(x) for x in range(10))
for _ in result:
    pass

# ✅ 正确：直接用 for 循环
for x in range(10):
    print(x)
python
# ❌ 生成器耗尽后继续使用
gen = (x for x in range(3))
print(list(gen))  # [0, 1, 2]
print(list(gen))  # [] ← 已耗尽！

# ✅ 正确：需要多次使用时用列表
data = [x for x in range(3)]
print(data)  # [0, 1, 2]
print(data)  # [0, 1, 2]
python
# ❌ 无限生成器没有终止条件
def infinite():
    while True:
        yield 1

for x in infinite():  # 无限循环！
    print(x)

# ✅ 正确：添加终止条件
for i, x in enumerate(infinite()):
    if i >= 100:
        break
    print(x)

列表推导式 vs 生成器表达式 ​

┌─────────────────────────────────────────────────────────────┐
│  列表推导式 vs 生成器表达式选择                               │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  列表推导式 [x for x in iterable]                           │
│  ─────────────────────────────────────                      │
│  • 立即生成所有元素                                         │
│  • 可以多次遍历                                             │
│  • 支持索引访问、len()                                      │
│  • 适用：小数据、需要多次访问、需要索引                      │
│                                                             │
│  生成器表达式 (x for x in iterable)                         │
│  ─────────────────────────────────────                      │
│  • 惰性生成，按需产出                                       │
│  • 只能遍历一次                                             │
│  • 不支持索引、len()                                        │
│  • 适用：大数据、流式处理、一次性使用                       │
│                                                             │
│  选择决策：                                                  │
│  数据量小 + 多次访问 → 列表推导式                            │
│  数据量大 + 一次性处理 → 生成器表达式                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

适用场景 ​

场景推荐方式原因
大数据集处理生成器表达式内存友好
文件逐行读取生成器函数流式处理
数据管道生成器函数惰性传递
无限序列生成器函数按需生成
小数据多次访问列表推导式支持重复遍历
需要索引访问列表推导式列表支持索引

L3 专家层：深入 ​

生成器内部结构 ​

python
import inspect

def simple_gen():
    x = 1
    yield x
    y = 2
    yield y
    return "done"

gen = simple_gen()

# 查看生成器属性
print(f"代码对象：{gen.gi_code}")
print(f"帧对象：{gen.gi_frame}")
print(f"运行状态：{gen.gi_running}")

# 帧对象保存局部变量
print(next(gen))
print(f"局部变量：{gen.gi_frame.f_locals}")  # {'x': 1}

print(next(gen))
print(f"局部变量：{gen.gi_frame.f_locals}")  # {'x': 1, 'y': 2}
┌─────────────────────────────────────────────────────────────┐
│  生成器内部结构                                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  generator 对象：                                            │
│  ├── gi_code  → 代码对象（编译后的字节码）                   │
│  ├── gi_frame → 帧对象（保存局部变量、执行位置）             │
│  ├── gi_running → 运行状态（是否正在执行）                  │
│  └── gi_yieldfrom → 委托的子生成器                          │
│                                                             │
│  帧对象 (Frame)：                                            │
│  ├── f_locals → 局部变量字典                                │
│  ├── f_lineno → 当前执行行号                                │
│  ├── f_lasti → 最后执行的指令索引                           │
│                                                             │
│  生成器状态：                                                │
│  GEN_CREATED  → 刚创建，未启动                               │
│  GEN_RUNNING  → 正在执行                                     │
│  GEN_SUSPENDED → 暂停在 yield                               │
│  GEN_CLOSED   → 已关闭                                       │
│                                                             │
└─────────────────────────────────────────────────────────────┘

生成器状态变化 ​

python
from inspect import getgeneratorstate

def state_demo():
    yield 1
    yield 2

gen = state_demo()
print(f"创建后：{getgeneratorstate(gen)}")  # GEN_CREATED

next(gen)
print(f"暂停后：{getgeneratorstate(gen)}")  # GEN_SUSPENDED

next(gen)
print(f"再暂停：{getgeneratorstate(gen)}")  # GEN_SUSPENDED

try:
    next(gen)
except StopIteration:
    pass
print(f"耗尽后：{getgeneratorstate(gen)}")  # GEN_CLOSED

yield 字节码 ​

python
import dis

def yield_demo():
    x = 1
    yield x
    x = x + 1
    yield x

dis.dis(yield_demo)
# 输出包含 YIELD_VALUE 指令
┌─────────────────────────────────────────────────────────────┐
│  yield 相关字节码                                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  YIELD_VALUE    → 暂停并返回值                               │
│  YIELD_FROM     → 委托给子生成器                            │
│  GET_YIELD_FROM_ITER → 获取 yield from 目标               │
│                                                             │
│  yield x 执行过程：                                          │
│  1. LOAD_FAST x  → 加载变量 x                               │
│  2. YIELD_VALUE  → 暂停，返回 x                             │
│  3. (下次恢复)   → 继续执行下一条指令                       │
│                                                             │
└─────────────────────────────────────────────────────────────┘

性能考量 ​

操作时间复杂度空间复杂度说明
创建生成器O(1)O(200字节)只分配对象和帧
yield 暂停/恢复O(1)O(1)保存/恢复帧状态
yield from 委托O(1)传递O(1)值直接传递
python
import timeit

list_time = timeit.timeit(lambda: sum([x for x in range(1_000_000)], number=10)
gen_time = timeit.timeit(lambda: sum(x for x in range(1_000_000)], number=10)

print(f"列表：{list_time:.3f}s")
print(f"生成器：{gen_time:.3f}s")

为什么 Python 选择 yield？ ​

┌─────────────────────────────────────────────────────────────┐
│  yield 设计动机                                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ① 简化迭代器实现                                            │
│  迭代器类需要 __iter__、__next__、状态管理                  │
│  生成器函数只需 yield，Python 自动处理协议                  │
│                                                             │
│  ② 惰性求值需求                                              │
│  大数据处理场景普遍                                         │
│  无限序列（斐波那契、素数）需要                              │
│  文件流、网络流处理                                         │
│                                                             │
│  ③ 协程基础                                                  │
│  yield 是协程的早期形式                                     │
│  async/await 基于类似机制                                   │
│  send() 实现双向通信                                        │
│                                                             │
│  ④ 代码可读性                                                │
│  yield 位置清晰标识暂停点                                   │
│  无需复杂的索引和状态变量                                   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

自检清单 ​

回答以下问题，检查你是否掌握了核心概念：

  1. gen = func() 会执行函数吗？
  2. yield 和 return 有什么区别？
  3. 生成器表达式和列表推导式有什么区别？
  4. 为什么生成器只能遍历一次？

答案：

  1. 不会，只创建生成器对象
  2. yield 暂停并返回多次，return 结束并返回一次
  3. 生成器表达式用圆括号、惰性；列表推导式用方括号、立即
  4. 生成器不存储数据，只维护执行状态，用完即弃

本章能力清单 ​

学完本章，你能够：

  • [x] 用 yield 创建生成器函数
  • [x] 理解 yield 的暂停/恢复机制
  • [x] 区分列表推导式和生成器表达式
  • [x] 用生成器处理大文件和数据流

下一步学习：

  • yield from 委托 → 第4节
  • itertools 高效处理 → 第5节

本章小结 ​

┌─────────────────────────────────────────────────────────────┐
│                    生成器基础 知识要点                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   核心概念：                                                 │
│   ✓ yield：暂停点打卡（返回值，暂停等待下次）               │
│   ✓ 生成器：惰性求值（内存友好）                            │
│                                                             │
│   执行流程：                                                 │
│   ✓ 创建时不执行                                            │
│   ✓ next() 开始执行                                         │
│   ✓ yield 暂停返回                                          │
│   ✓ 函数结束 StopIteration                                 │
│                                                             │
│   关键区别：                                                 │
│   ✓ 列表推导式 [x for x] - 立即生成                         │
│   ✓ 生成器表达式 (x for x) - 惰性生成                       │
│                                                             │
│   应用场景：                                                 │
│   ✓ 大数据处理                                              │
│   ✓ 文件逐行读取                                            │
│   ✓ 无限序列                                                │
│   ✓ 数据管道                                                │
│                                                             │
└─────────────────────────────────────────────────────────────┘