【Python】第 7 章：生成器与协程

第 7 章生成器与协程7.1 迭代器协议原理讲解迭代器协议Iterator Protocol┌─────────────────────────────────────────────────────────┐ │ Python 迭代器协议 │ ├─────────────────────────────────────────────────────────┤ │ │ │ 可迭代对象 (Iterable): │ │ - 实现 __iter__() 方法 │ │ - 返回迭代器对象 │ │ - 例如list, tuple, dict, str, set │ │ │ │ 迭代器 (Iterator): │ │ - 实现 __iter__() 方法 (返回自身) │ │ - 实现 __next__() 方法 (返回下一个值) │ │ - 耗尽时抛出 StopIteration 异常 │ │ │ │ 协议流程 │ │ ┌─────────────────────────────────────────────────┐ │ │ │ for item in iterable: │ │ │ │ process(item) │ │ │ │ │ │ │ │ 内部实现 │ │ │ │ iterator iterable.__iter__() │ │ │ │ while True: │ │ │ │ try: │ │ │ │ item iterator.__next__() │ │ │ │ process(item) │ │ │ │ except StopIteration: │ │ │ │ break │ │ │ └─────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────┘手动实现迭代器classCountDown:倒计时迭代器def__init__(self,start):self.startstartdef__iter__(self):returnselfdef__next__(self):ifself.start0:raiseStopIteration self.start-1returnself.start1# 使用foriinCountDown(5):print(i)# 5, 4, 3, 2, 1生成器函数生成器 自动实现迭代器协议# 普通函数defcountdown_normal(n):result[]foriinrange(n,0,-1):result.append(i)returnresult# 一次性返回所有值# 生成器函数defcountdown_generator(n):foriinrange(n,0,-1):yieldi# 每次返回一个值暂停生成器特点┌─────────────────────────────────────────────────────────┐ │ 生成器 vs 普通函数 │ ├─────────────────────────────────────────────────────────┤ │ │ │ 普通函数 │ │ ┌─────────────────────────────────────────────────┐ │ │ │ def func(): │ │ │ │ return 1 │ │ │ │ return 2 # 永远不会执行 │ │ │ │ │ │ │ │ 调用返回结果函数结束 │ │ │ │ 内存所有结果在内存中 │ │ │ └─────────────────────────────────────────────────┘ │ │ │ │ 生成器函数 │ │ ┌─────────────────────────────────────────────────┐ │ │ │ def gen(): │ │ │ │ yield 1 # 暂停保存状态 │ │ │ │ yield 2 # 恢复继续执行 │ │ │ │ │ │ │ │ 调用返回生成器对象不执行代码 │ │ │ │ 内存惰性计算节省内存 │ │ │ └─────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────┘yield 的工作原理生成器状态机┌─────────────────────────────────────────────────────────┐ │ yield 状态转换 │ ├─────────────────────────────────────────────────────────┤ │ │ │ 生成器状态 │ │ ┌─────────────┐ │ │ │ CREATED │ 刚创建未执行 │ │ └──────┬──────┘ │ │ │ next() │ │ ↓ │ │ ┌─────────────┐ │ │ │ RUNNING │ 正在执行 │ │ └──────┬──────┘ │ │ │ yield │ │ ↓ │ │ ┌─────────────┐ │ │ │ SUSPENDED │ 暂停保存局部变量 │ │ └──────┬──────┘ │ │ │ next() │ │ ↓ │ │ ┌─────────────┐ │ │ │ RUNNING │ 从 yield 处恢复 │ │ └──────┬──────┘ │ │ │ return / 结束 │ │ ↓ │ │ ┌─────────────┐ │ │ │ CLOSED │ 抛出 StopIteration │ │ └─────────────┘ │ │ │ │ 生成器对象属性 │ │ - gi_frame: 栈帧对象 │ │ - gi_running: 是否正在运行 │ │ - gi_code: 代码对象 │ │ - gi_yieldfrom: yield from 的目标 │ │ │ └─────────────────────────────────────────────────────────┘查看生成器状态defgen():x1yieldx x2yieldx ggen()print(f创建后gi_running{g.gi_running})# Falsenext(g)print(f第一次 yield 后gi_running{g.gi_running})# Falsenext(g)print(f第二次 yield 后gi_running{g.gi_running})# Falsetry:next(g)exceptStopIteration:print(生成器已结束)7.2 协程基础async/await 原理协程Coroutine可暂停的函数┌─────────────────────────────────────────────────────────┐ │ async/await 原理 │ ├─────────────────────────────────────────────────────────┤ │ │ │ async 函数 │ │ - 定义协程函数 │ │ - 调用返回协程对象不执行 │ │ - 需要 await 或事件循环驱动 │ │ │ │ await 表达式 │ │ - 暂停当前协程 │ │ - 等待可等待对象awaitable完成 │ │ - 恢复时获取结果 │ │ │ │ 可等待对象 │ │ - 协程对象 │ │ - Task 对象 │ │ - 实现 __await__() 的对象 │ │ │ │ 执行流程 │ │ ┌─────────────────────────────────────────────────┐ │ │ │ async def fetch(): │ │ │ │ result await http.get() # 暂停 │ │ │ │ return result │ │ │ │ │ │ │ │ # 驱动协程 │ │ │ │ asyncio.run(fetch()) │ │ │ └─────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────┘协程状态机importasyncioimportinspectasyncdefcoro():awaitasyncio.sleep(0.1)returndone# 创建协程对象coro_objcoro()# 检查类型print(f是协程{inspect.iscoroutine(coro_obj)})# True# 查看状态Python 3.8# print(f状态{coro_obj.cr_running})# 运行协程resultasyncio.run(coro_obj)print(f结果{result})事件循环事件循环Event Loop工作原理┌─────────────────────────────────────────────────────────┐ │ 事件循环架构 │ ├─────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ 事件循环 (Event Loop) │ │ │ │ ┌─────────────────────────────────────────┐ │ │ │ │ │ while True: │ │ │ │ │ │ # 1. 运行就绪的回调 │ │ │ │ │ │ run_ready_callbacks() │ │ │ │ │ │ │ │ │ │ │ │ # 2. 处理 I/O 事件 │ │ │ │ │ │ process_io_events() │ │ │ │ │ │ │ │ │ │ │ │ # 3. 调度定时任务 │ │ │ │ │ │ schedule_timers() │ │ │ │ │ │ │ │ │ │ │ │ # 4. 如果没有任务等待 │ │ │ │ │ │ if no_tasks: │ │ │ │ │ │ wait_for_events() │ │ │ │ │ └─────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────┘ │ │ ↕ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ 任务队列 │ │ │ │ [Task1, Task2, Task3, ...] │ │ │ │ │ │ │ │ 每个 Task 包装一个协程 │ │ │ └─────────────────────────────────────────────────┘ │ │ ↕ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ I/O 多路复用 │ │ │ │ - select (跨平台) │ │ │ │ - epoll (Linux) │ │ │ │ - kqueue (BSD/macOS) │ │ │ │ - IOCP (Windows) │ │ │ └─────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────┘asyncio 基础核心组件importasyncio# 1. 创建事件循环loopasyncio.get_event_loop()# 2. 运行协程asyncio.run(main())# Python 3.7# 3. 创建任务taskasyncio.create_task(coro())# 4. 等待多个任务resultsawaitasyncio.gather(task1,task2,task3)# 5. 超时控制try:resultawaitasyncio.wait_for(coro(),timeout5.0)exceptasyncio.TimeoutError:print(超时)# 6. 并发限制semaphoreasyncio.Semaphore(10)asyncwithsemaphore:awaitdo_something()7.3 实践实现一个简单的协程实验代码# examples/chapter-07/generator_coroutine_examples.pyimportasyncioimporttimefromtypingimportGeneratorprint(*70)print(生成器与协程实践示例)print(*70)# 1. 生成器基础 print(\n【示例 1】生成器基础)print(-*50)defsimple_generator():简单生成器yield1yield2yield3gensimple_generator()print(f第一次 next:{next(gen)})print(f第二次 next:{next(gen)})print(f第三次 next:{next(gen)})# 使用 for 循环print(\n使用 for 循环:)forvalueinsimple_generator():print(f 值{value})# 2. 生成器发送数据 print(\n【示例 2】生成器发送数据 (send))print(-*50)defaccumulator():累加器生成器total0whileTrue:valueyieldtotalifvalueisNone:breaktotalvalue accaccumulator()print(f初始值{next(acc)})# 启动生成器print(f发送 5:{acc.send(5)})print(f发送 3:{acc.send(3)})print(f发送 10:{acc.send(10)})acc.send(None)# 关闭# 3. yield from print(\n【示例 3】yield from 委托生成)print(-*50)defsub_generator():yieldsub-1yieldsub-2returnsub-returndefmain_generator():yieldmain-1resultyieldfromsub_generator()yieldfsub returned:{result}yieldmain-2forvalueinmain_generator():print(f{value})# 4. 生成器实现管道 print(\n【示例 4】生成器管道)print(-*50)defnumbers():生成数字foriinrange(10):yieldidefsquares(seq):计算平方forninseq:yieldn**2defevens(seq):过滤偶数forninseq:ifn%20:yieldn# 管道组合pipelineevens(squares(numbers()))print(f偶数平方{list(pipeline)})# 5. 异步协程基础 print(\n【示例 5】异步协程基础)print(-*50)asyncdefasync_hello():异步 helloprint( 协程开始)awaitasyncio.sleep(0.1)print( 协程结束)returnhello# 运行协程resultasyncio.run(async_hello())print(f结果{result})# 6. 并发执行多个协程 print(\n【示例 6】并发执行多个协程)print(-*50)asyncdeffetch_data(name,delay):模拟获取数据print(f{name}开始 (延迟{delay}s))awaitasyncio.sleep(delay)print(f{name}完成)returnf{name}的数据asyncdefconcurrent_fetch():tasks[fetch_data(A,0.3),fetch_data(B,0.2),fetch_data(C,0.1),]# 并发执行starttime.time()resultsawaitasyncio.gather(*tasks)elapsedtime.time()-startprint(f\n总耗时{elapsed:.2f}s (如果串行需要 0.6s))print(f结果{results})asyncio.run(concurrent_fetch())# 7. 异步上下文管理器 print(\n【示例 7】异步上下文管理器)print(-*50)classAsyncResource:异步资源示例asyncdef__aenter__(self):print( 获取资源...)awaitasyncio.sleep(0.1)print( 资源已获取)returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):print( 释放资源...)awaitasyncio.sleep(0.1)print( 资源已释放)asyncdefdo_work(self):print( 工作中...)awaitasyncio.sleep(0.1)asyncdefuse_resource():asyncwithAsyncResource()asres:awaitres.do_work()asyncio.run(use_resource())# 8. 生成器 vs 协程性能对比 print(\n【示例 8】生成器 vs 列表性能)print(-*50)defgen_range(n):foriinrange(n):yieldideflist_range(n):returnlist(range(n))n1000000# 生成器内存importsys gen_objgen_range(n)print(f生成器内存{sys.getsizeof(gen_obj)}bytes)# 列表内存list_objlist_range(1000)# 小一点的列表print(f列表内存 (1000 个):{sys.getsizeof(list_obj)}bytes)print(\n生成器优势惰性计算节省内存)print(列表优势可重复迭代支持索引)实验练习练习 1实现范围生成器defmy_range(start,stopNone,step1):手动实现 range 生成器ifstopisNone:stopstart start0currentstartwhile(step0andcurrentstop)or(step0andcurrentstop):yieldcurrent currentstep# 测试print(my_range(5):,list(my_range(5)))print(my_range(2, 8):,list(my_range(2,8)))print(my_range(0, 10, 2):,list(my_range(0,10,2)))print(my_range(10, 0, -1):,list(my_range(10,0,-1)))练习 2实现异步生产者 - 消费者importasyncioimportrandomasyncdefproducer(queue,name,count):生产者foriinrange(count):itemf{name}-{i}awaitqueue.put(item)print(f生产{item})awaitasyncio.sleep(random.uniform(0.01,0.05))awaitqueue.put(None)# 结束标记asyncdefconsumer(queue,name):消费者whileTrue:itemawaitqueue.get()ifitemisNone:breakprint(f{name}消费{item})awaitasyncio.sleep(random.uniform(0.02,0.08))queue.task_done()asyncdefmain():queueasyncio.Queue(maxsize5)# 创建生产者和消费者producers[producer(queue,P1,10),producer(queue,P2,10),]consumers[consumer(queue,C1),consumer(queue,C2),]# 并发运行awaitasyncio.gather(*producers,*consumers)asyncio.run(main())练习 3使用生成器处理大文件defread_lines(filename):惰性读取文件行withopen(filename,r)asf:forlineinf:yieldline.strip()deffilter_non_empty(lines):过滤空行forlineinlines:ifline:yieldlinedeftransform_upper(lines):转换为大写forlineinlines:yieldline.upper()# 创建处理管道defprocess_file(filename):linesread_lines(filename)linesfilter_non_empty(lines)linestransform_upper(lines)forlineinlines:print(line)# 创建测试文件withopen(/tmp/test.txt,w)asf:f.write(hello\n\nworld\n\npython\n)# 处理print(处理文件:)process_file(/tmp/test.txt)常见问题Q1: 生成器和协程有什么区别A:特性生成器协程定义defyieldasync defawait驱动next()/send()事件循环用途迭代、管道异步 I/O、并发返回值yieldreturnawaitQ2: 为什么协程需要事件循环A:协程不会自动执行事件循环调度和驱动协程管理 I/O 多路复用处理并发和任务调度Q3:yield和yield from的区别A:# yield: 返回单个值yieldvalue# yield from: 委托给另一个迭代器yieldfromiterable# 等价于 for item in iterable: yield itemQ4: 如何关闭生成器A:gengenerator()next(gen)# 方法 1: 发送 Nonegen.send(None)# 方法 2: 调用 close()gen.close()# 方法 3: 抛出异常gen.throw(StopIteration)Q5: async/await 和 threading 的区别A:asyncio: 单线程协作式多任务适合 I/O 密集threading: 多线程抢占式多任务适合 CPU 密集但有 GIL 限制multiprocessing: 多进程真正并行适合 CPU 密集本章小结迭代器协议__iter__()__next__()生成器自动实现迭代器协议使用yield生成器惰性计算节省内存yield from委托给子生成器协程使用async/await语法事件循环驱动协程执行asyncio 提供异步 I/O 基础设施下一章预告第 8 章将深入探讨GIL 与并发包括全局解释器锁、多线程和多进程的对比。