Python yield 用法大全

yield 是 Python 中的关键字，用于定义生成器函数。与 return 不同，yield 会使函数返回一个生成器对象，这个对象可以迭代，但不会立即执行函数体。每次调用生成器的 next() 方法时，函数会从上次暂停的地方继续执行，直到遇到下一个 yield 语句。

1. 生成器 vs 普通函数

• 普通函数：使用 return，一次性返回所有结果，函数执行结束
• 生成器函数：使用 yield，可以多次返回值，每次返回后暂停，保持状态

2. 核心特性

• 惰性求值：只在需要时生成值，节省内存
• 状态保持：局部变量和执行状态在 yield 之间保持不变
• 可迭代：生成器是迭代器，可以用在 for 循环中
• 无限序列：可以表示无限长的序列（如无限流数据）

3. 用法

3.1 创建简单生成器

代码示例：

defbasic_generator():"""基本生成器示例"""    print("开始执行")yield1    print("继续执行")yield2    print("结束执行")yield3# 使用方式gen = basic_generator()print("第一次调用:", next(gen))  # 输出: 开始执行 → 1print("第二次调用:", next(gen))  # 输出: 继续执行 → 2print("第三次调用:", next(gen))  # 输出: 结束执行 → 3# 或者使用 for 循环for value in basic_generator():    print("循环获取:", value)

3.2 循环中使用 yield

代码示例：

deffibonacci_limited(n):"""生成前n个斐波那契数"""    a, b = 0, 1    count = 0while count < n:yield a        a, b = b, a + b        count += 1# 使用fib = fibonacci_limited(10)print(list(fib))  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]# 或者直接迭代for num in fibonacci_limited(5):    print(num, end=' ')  # 0 1 1 2 3

3.3  yield from（Python 3.3+）

yield from 用于在生成器中委托给另一个生成器或可迭代对象，可以简化代码并提高效率。

• 用于扁平化嵌套的生成器
• 自动处理子生成器的异常
• 可以接收子生成器的返回值

代码示例：

defchain_generators():"""连接多个生成器"""yieldfrom range(3)          # 委托给 rangeyieldfrom (x*2for x in range(3))  # 委托给生成器表达式yieldfrom ['a', 'b', 'c']   # 委托给列表print(list(chain_generators()))  # 输出: [0, 1, 2, 0, 2, 4, 'a', 'b', 'c']# 嵌套委托示例defflatten(nested_list):"""扁平化嵌套列表"""for sublist in nested_list:if isinstance(sublist, list):yieldfrom flatten(sublist)else:yield sublistnested = [1, [2, [3, 4], 5], 6]print(list(flatten(nested)))  # [1, 2, 3, 4, 5, 6]

3.4 生成器表达式

生成器表达式是创建生成器的简洁语法，类似于列表推导式，但使用圆括号。

• 语法：(expression for item in iterable if condition)
• 惰性求值，不立即创建完整列表
• 内存效率高

代码示例：

# 基本生成器表达式squares = (x**2for x in range(10))print(next(squares))  # 0print(next(squares))  # 1print(list(squares))  # [4, 9, 16, 25, 36, 49, 64, 81]# 带条件的生成器表达式even_squares = (x**2for x in range(10) if x % 2 == 0)print(list(even_squares))  # [0, 4, 16, 36, 64]# 多层循环pairs = ((x, y) for x in range(3) for y in range(3))print(list(pairs))  # [(0,0), (0,1), (0,2), (1,0), (1,1), (1,2), (2,0), (2,1), (2,2)]# 管道式处理numbers = range(100)pipeline = (    x * 2# 步骤1: 乘以2for x in numbers         # 输入if x % 3 == 0# 步骤2: 过滤3的倍数if x < 50# 步骤3: 只取小于50的)print(list(pipeline)[:5])  # [0, 6, 12, 18, 24]

3.5 协程（Coroutine）

生成器可以用于实现简单的协程，支持双向通信。

• 使用 .send(value) 向生成器发送数据
• 使用 .throw(exception) 向生成器抛出异常
• 使用 .close() 关闭生成器
• yield 可以接收外部发送的值

代码示例：

defcoroutine_example():"""简单的协程示例"""    print("协程启动")try:whileTrue:            received = yield# 接收外部发送的值            print(f"收到: {received}")except GeneratorExit:        print("协程关闭")except ValueError as e:        print(f"捕获异常: {e}")yield"处理了异常"# 可以继续yield# 使用协程coro = coroutine_example()next(coro)              # 启动协程，输出: "协程启动"coro.send("Hello")      # 输出: "收到: Hello"coro.send(123)          # 输出: "收到: 123"# 抛出异常coro.throw(ValueError("测试异常"))  # 输出: "捕获异常: 测试异常"print(next(coro))                   # 输出: "处理了异常"coro.close()            # 输出: "协程关闭"# 带初始值的协程defaccumulator():"""累加器协程"""    total = 0whileTrue:        value = yield totalif value isNone:break        total += valueacc = accumulator()next(acc)               # 启动print(acc.send(10))     # 输出: 10print(acc.send(20))     # 输出: 30print(acc.send(5))      # 输出: 35

3.6 生成器作为上下文管理器

生成器可以和 contextlib.contextmanager 装饰器结合，创建自定义上下文管理器。

• 使用 @contextmanager 装饰器
• yield 之前的部分相当于 __enter__
• yield 之后的部分相当于 __exit__

代码示例：

from contextlib import contextmanager@contextmanagerdeftimer():"""计时上下文管理器"""import time    start = time.time()try:yield# 这里可以yield一个值给as变量finally:        end = time.time()        print(f"耗时: {end - start:.2f}秒")# 使用with timer():    sum(range(1000000))  # 输出: 耗时: 0.02秒# 带返回值的上下文管理器@contextmanagerdefopen_file(filename, mode='r'):"""文件操作的上下文管理器"""    file = open(filename, mode)try:yield file  # 将文件对象交给用户finally:        file.close()with open_file('test.txt', 'w') as f:    f.write("Hello, World!")

4、应用举例

4.1 处理大文件

代码示例：

defread_large_file(file_path, chunk_size=1024):"""分块读取大文件"""with open(file_path, 'r', encoding='utf-8') as f:whileTrue:            chunk = f.read(chunk_size)ifnot chunk:breakyield chunkdefprocess_log_file(file_path):"""逐行处理日志文件，过滤包含ERROR的行"""with open(file_path, 'r') as f:for line in f:if'ERROR'in line:yield line.strip()# 使用for error_line in process_log_file('app.log'):    print(f"发现错误: {error_line}")

4.2 无限序列

代码示例：

definfinite_counter(start=0):"""无限计数器"""whileTrue:yield start        start += 1definfinite_random():"""无限随机数生成器"""import randomwhileTrue:yield random.random()# 限制使用counter = infinite_counter(10)for _ in range(5):    print(next(counter), end=' ')  # 10 11 12 13 14# 使用itertools.islice限制from itertools import islicerandom_gen = infinite_random()first_5 = list(islice(random_gen, 5))print(first_5)

4.3 数据管道和流处理

代码示例：

defdata_pipeline(data):"""数据处理管道"""# 第一阶段：过滤    filtered = (item for item in data if item > 0)# 第二阶段：转换    transformed = (item * 2for item in filtered)# 第三阶段：分组yieldfrom transformed# 复杂管道示例defetl_pipeline(data_stream):"""ETL管道：提取、转换、加载"""# 提取：从原始数据流    raw_data = (item for item in data_stream)# 转换：清理和转换数据deftransform():for item in raw_data:# 清理数据            cleaned = str(item).strip().lower()if cleaned:  # 跳过空值yield cleaned# 加载：批处理    batch_size = 100    batch = []for item in transform():        batch.append(item)if len(batch) >= batch_size:yield batch            batch = []if batch:  # 处理最后一批yield batch# 模拟数据流test_data = ["  Hello  ", "  WORLD  ", "", "  Python  ", None, "  Generator  "]for batch in etl_pipeline(test_data):    print(f"处理批次: {batch}")

4.4 状态机实现

代码示例：

defstate_machine():"""简单的状态机"""    state = "START"whileTrue:if state == "START":            print("状态: START")            command = yieldif command == "go":                state = "RUNNING"elif state == "RUNNING":            print("状态: RUNNING")            command = yieldif command == "stop":                state = "STOPPED"elif command == "pause":                state = "PAUSED"elif state == "PAUSED":            print("状态: PAUSED")            command = yieldif command == "resume":                state = "RUNNING"elif command == "stop":                state = "STOPPED"elif state == "STOPPED":            print("状态: STOPPED")break# 使用状态机sm = state_machine()next(sm)                # 启动sm.send("go")           # 切换到RUNNINGsm.send("pause")        # 切换到PAUSEDsm.send("resume")       # 切回RUNNINGsm.send("stop")         # 切换到STOPPED

5. 性能优化

1. 内存对比

代码示例：

import sysimport timedeftest_memory_performance():"""测试内存和性能对比"""    n = 1000000# 方法1：使用列表（占用大量内存）    print("方法1: 使用列表")    start = time.time()    lst = [x**2for x in range(n)]    memory_list = sys.getsizeof(lst)    time_list = time.time() - start    print(f"  内存: {memory_list:,} 字节")    print(f"  时间: {time_list:.3f} 秒")# 方法2：使用生成器（内存高效）    print("\n方法2: 使用生成器")    start = time.time()    gen = (x**2for x in range(n))    memory_gen = sys.getsizeof(gen)# 消费生成器    sum_gen = sum(gen)    time_gen = time.time() - start    print(f"  内存: {memory_gen:,} 字节")    print(f"  时间: {time_gen:.3f} 秒")    print(f"  总和: {sum_gen:,}")    print(f"\n内存节省: {(memory_list - memory_gen) / memory_list * 100:.1f}%")test_memory_performance()

2. 惰性求值优化

代码示例：

deflazy_evaluation_demo():"""惰性求值优化示例"""# 过早求值（不推荐）defprocess_data_eager(data):# 这里立即创建了所有中间列表        filtered = [x for x in data if x > 0]  # 中间列表1        squared = [x**2for x in filtered]     # 中间列表2return sum(squared)                    # 最终结果# 惰性求值（推荐）defprocess_data_lazy(data):# 使用生成器表达式，没有中间列表        filtered = (x for x in data if x > 0)        squared = (x**2for x in filtered)return sum(squared)# 测试import random    data = [random.randint(-100, 100) for _ in range(1000000)]import time    start = time.time()    result1 = process_data_eager(data)    time1 = time.time() - start    start = time.time()    result2 = process_data_lazy(data)    time2 = time.time() - start    print(f"过早求值: 结果={result1:,}, 时间={time1:.3f}秒")    print(f"惰性求值: 结果={result2:,}, 时间={time2:.3f}秒")    print(f"速度提升: {time1/time2:.1f}倍")lazy_evaluation_demo()

6. 注意事项

代码示例：

defcommon_mistakes():"""常见错误示例"""# 错误1：重复使用已耗尽的生成器    gen = (x for x in range(3))    print("错误1 - 重复使用生成器:")    print(f"  第一次: {list(gen)}")  # [0, 1, 2]    print(f"  第二次: {list(gen)}")  # [] - 空了！# 错误2：修改外部变量defbad_generator():        results = []for i in range(3):            results.append(i)yield results  # 每次都返回同一个列表的引用    print("\n错误2 - 返回可变对象引用:")    bg = bad_generator()for item in bg:        print(f"  {item}")  # 每次都是[0, 1, 2]，而不是[0], [0,1], [0,1,2]# 正确做法：返回副本或不可变对象defgood_generator():        results = []for i in range(3):            results.append(i)yield results.copy()  # 返回副本    print("\n正确做法 - 返回副本:")    gg = good_generator()for item in gg:        print(f"  {item}")  # [0], [0,1], [0,1,2]common_mistakes()

7. 特殊用法

1. 递归生成器

代码示例：

defrecursive_generator(n):"""递归生成器"""if n <= 0:yield0else:yield nyieldfrom recursive_generator(n-1)print(list(recursive_generator(5)))  # [5, 4, 3, 2, 1, 0]# 遍历树结构classTreeNode:def__init__(self, value, left=None, right=None):        self.value = value        self.left = left        self.right = rightdefinorder_traversal(node):"""中序遍历二叉树"""if node:yieldfrom inorder_traversal(node.left)yield node.valueyieldfrom inorder_traversal(node.right)# 构建示例树tree = TreeNode(1,                TreeNode(2,                         TreeNode(4),                         TreeNode(5)),                TreeNode(3,                         TreeNode(6),                         TreeNode(7)))print(list(inorder_traversal(tree)))  # [4, 2, 5, 1, 6, 3, 7]

2. 异步生成器（Python 3.6+）

代码示例：

import asyncioasyncdefasync_generator():"""异步生成器"""for i in range(5):await asyncio.sleep(0.1)  # 模拟异步操作yield iasyncdefuse_async_generator():"""使用异步生成器"""asyncfor value in async_generator():        print(f"异步获取: {value}")# 运行asyncio.run(use_async_generator())

总结

yield 是 Python 中非常强大的特性，主要用途包括：

1. 创建生成器：实现惰性求值，节省内存
2. 实现协程：支持双向通信的轻量级并发
3. 构建数据管道：处理流式数据
4. 创建无限序列：表示潜在的无限数据流
5. 实现状态机：保持状态的可迭代对象

掌握这些用法，有助于我们写出更高效、更 Pythonic 的代码。