[Python3高阶编程] - 深入理解yield 二: 假如没有yield怎么写代码？

1. 没有yield时的核心问题1.1 内存爆炸问题Memory Explosion场景处理大型数据集# 没有 yield 的情况下 - 必须一次性加载所有数据 def get_all_users(): 获取所有用户信息 users [] # 假设有 1000 万用户 for i in range(10_000_000): user fetch_user_from_database(i) # 从数据库获取用户 users.append(user) return users # 返回包含 1000 万个对象的列表 # 使用时 all_users get_all_users() # 内存瞬间占用巨大 for user in all_users: process_user(user)问题需要等待很长时间才能开始处理第一个用户内存可能直接溢出OOM - Out of Memory如果只需要前100个用户也必须加载全部1000万个有了 yield 的解决方案def get_all_users(): 使用 yield - 按需生成 for i in range(10_000_000): user fetch_user_from_database(i) yield user # 立即返回不存储在内存中 # 使用时 users get_all_users() for i, user in enumerate(users): process_user(user) if i 99: # 只需要前100个 break # 循环结束不会继续查询数据库1.2 无法表示无限序列没有 yield 的困境# 如何表示自然数序列不可能 # def natural_numbers(): # return [1, 2, 3, 4, ...] # 这是不可能的 # 只能用有限的方式模拟 def get_first_n_numbers(n): return list(range(1, n1)) # 如果不知道需要多少个数就很麻烦 numbers_needed calculate_how_many_needed() # 运行时才知道 numbers get_first_n_numbers(numbers_needed)有了 yield 的优雅解决方案def natural_numbers(): n 1 while True: yield n n 1 # 使用时完全灵活 def find_first_prime_over_million(): for num in natural_numbers(): if num 1_000_000 and is_prime(num): return num1.3 数据处理管道复杂化没有 yield 时的数据处理# 处理一个大文件的每一行 def process_large_file(filename): # 第一步读取所有行到内存 all_lines [] with open(filename) as f: for line in f: all_lines.append(line.strip()) # 第二步过滤空行 non_empty_lines [] for line in all_lines: if line: non_empty_lines.append(line) # 第三步转换为大写 upper_lines [] for line in non_empty_lines: upper_lines.append(line.upper()) # 第四步添加前缀 prefixed_lines [] for line in upper_lines: prefixed_lines.append(fPROCESSED: {line}) return prefixed_lines # 所有中间结果都占用内存内存使用情况原始数据1GB过滤后800MB转换后800MB最终结果900MB总内存峰值1GB 800MB 800MB 900MB 3.5GB有了 yield 的高效处理def read_lines(filename): with open(filename) as f: for line in f: yield line.strip() def filter_empty(lines): for line in lines: if line: yield line def to_upper(lines): for line in lines: yield line.upper() def add_prefix(lines): for line in lines: yield fPROCESSED: {line} # 组合使用 def process_large_file(filename): lines read_lines(filename) lines filter_empty(lines) lines to_upper(lines) lines add_prefix(lines) return lines # 返回生成器内存使用恒定 # 内存使用始终只有当前处理的一行数据2. 在 yield 出现之前开发者如何解决这些问题2.1 回调函数模式Callback Pattern# 模拟没有 yield 的时代的做法 def process_data_with_callback(data_source, callback): 通过回调函数处理每个数据项 for item in data_source: processed_item process_item(item) callback(processed_item) # 立即处理不存储 # 使用 def handle_item(item): print(f处理: {item}) process_data_with_callback(large_dataset, handle_item)问题代码分散逻辑不连贯难以组合多个处理步骤错误处理复杂2.2 迭代器类Iterator Classes# 手动实现迭代器 class NumberIterator: def __init__(self, start, end): self.current start self.end end def __iter__(self): return self def __next__(self): if self.current self.end: raise StopIteration value self.current self.current 1 return value # 使用 numbers NumberIterator(1, 1000000) for num in numbers: print(num)问题代码冗长样板代码多每个新需求都要写一个完整的类状态管理复杂容易出错2.3 分块处理Chunkingdef process_in_chunks(data_source, chunk_size1000): 分块处理数据 chunk [] for item in data_source: chunk.append(item) if len(chunk) chunk_size: process_chunk(chunk) chunk [] if chunk: process_chunk(chunk)问题仍然需要临时存储 chunk_size 的数据chunk_size 的选择很困难无法做到真正的逐项处理3. yield 是如何彻底解决这些问题的3.1 协程机制Coroutine Mechanismyield实际上创建了一个协程它具有以下特性def demonstrate_coroutine_state(): print(1. 函数开始执行) x 10 print(f2. x {x}) yield x print(3. 恢复执行) x 5 print(f4. x {x}) yield x print(5. 再次恢复执行) x * 2 print(f6. x {x}) yield x # 测试状态保存 gen demonstrate_coroutine_state() print(第一次调用:) val1 next(gen) # 输出1,2返回10 print(第二次调用:) val2 next(gen) # 输出3,4返回15 print(第三次调用:) val3 next(gen) # 输出5,6返回30关键突破自动状态保存局部变量、执行位置都被自动保存暂停/恢复机制函数可以暂停并在稍后恢复简洁语法几行代码就能实现复杂的迭代逻辑3.2 内存效率的革命性提升# 对比处理1亿个数字 import sys # 方法1列表没有yield的情况 def create_list(): return [i for i in range(100_000_000)] # 方法2生成器有yield的情况 def create_generator(): for i in range(100_000_000): yield i # 内存对比 # list_obj create_list() # 需要 ~800MB 内存 # gen_obj create_generator() # 只需要 ~100字节内存 print(生成器创建瞬间完成列表创建需要几秒钟和大量内存)3.3 统一的迭代协议yield让任何函数都能轻松成为迭代器# 在 yield 出现前要实现迭代器需要 class CustomIterator: def __init__(self, data): self.data data self.index 0 def __iter__(self): return self def __next__(self): if self.index len(self.data): raise StopIteration value self.data[self.index] self.index 1 return value # 有了 yield 后 def custom_iterator(data): for item in data: yield item # 功能完全相同但代码简洁了80%4. 实际案例Web 开发中的文件上传没有 yield 的问题# 传统方式处理大文件上传 def handle_file_upload(request): # 1. 接收整个文件到内存 file_content request.body # 假设是1GB的大文件 # 2. 处理文件比如压缩、转换格式等 processed_content process_entire_file(file_content) # 3. 保存到磁盘 save_to_disk(processed_content) return 上传成功问题服务器内存被大量占用多个用户同时上传会导致内存耗尽用户要等很久才知道上传是否成功有了 yield 的解决方案def stream_file_processing(request): 流式处理文件上传 buffer bytearray() # 逐块读取请求体 for chunk in request.stream: buffer.extend(chunk) # 当缓冲区足够大时处理并清空 if len(buffer) 64 * 1024: # 64KB processed_chunk process_chunk(buffer) yield processed_chunk # 立即发送处理结果 buffer.clear() # 处理剩余数据 if buffer: yield process_chunk(buffer) # 在Web框架中使用如Flask app.route(/upload, methods[POST]) def upload(): return Response( stream_file_processing(request), mimetypeapplication/octet-stream )优势内存使用恒定只缓存64KB用户可以立即看到进度支持更大的文件上传服务器可以同时处理更多请求5. 总结yield 带来的根本性改变问题类型没有 yield 的困境有了 yield 的解决方案内存效率必须一次性加载所有数据按需生成内存恒定无限序列无法表示轻松实现无限生成器代码复杂度需要手动管理状态和迭代自动状态保存简洁语法数据管道多个中间列表内存浪费链式生成器零额外内存响应速度必须等待全部数据准备完成立即开始处理第一个元素yield不仅仅是一个语法糖它是 Python 对惰性求值和协程概念的优雅实现从根本上改变了我们处理数据流的方式使得 Python 在处理大数据、构建数据管道、实现并发等方面变得更加高效和简洁。