Python 生成器与协程异步编程实战核心概念与原理Python 中的生成器和协程是实现异步编程的重要工具它们为处理 I/O 密集型任务提供了高效的解决方案。生成器Generators生成器是一种特殊的迭代器通过yield语句实现可以逐个生成值而不是一次性生成所有值从而节省内存。协程Coroutines协程是可以暂停执行并在将来恢复的函数通过async和await关键字实现是 Python 3.5 中异步编程的核心。生成器的实现与应用基本生成器# 基本生成器函数 def simple_generator(): yield 1 yield 2 yield 3 # 使用生成器 for value in simple_generator(): print(value) # 输出: 1, 2, 3生成器表达式# 生成器表达式 gen (x * 2 for x in range(5)) for value in gen: print(value) # 输出: 0, 2, 4, 6, 8高级生成器应用# 无限序列生成器 def fibonacci(): a, b 0, 1 while True: yield a a, b b, a b # 使用无限生成器 fib fibonacci() for i in range(10): print(next(fib)) # 输出: 0, 1, 1, 2, 3, 5, 8, 13, 21, 34协程的实现与应用基本协程import asyncio # 基本协程函数 async def hello(): print(Hello) await asyncio.sleep(1) print(World) # 运行协程 asyncio.run(hello()) # 输出: # Hello # (等待1秒) # World并发协程import asyncio import time async def task(name, delay): print(fTask {name} started) await asyncio.sleep(delay) print(fTask {name} completed) return fResult from {name} async def main(): # 并发执行多个协程 results await asyncio.gather( task(A, 2), task(B, 1), task(C, 3) ) print(results) asyncio.run(main()) # 输出: # Task A started # Task B started # Task C started # Task B completed # Task A completed # Task C completed # [Result from A, Result from B, Result from C]性能分析内存使用对比方法内存使用 (MB)适用场景列表100数据量小需要随机访问生成器1数据量大顺序访问协程1I/O 密集型任务执行时间对比import time import asyncio # 测试同步执行 def sync_task(name, delay): print(fSync Task {name} started) time.sleep(delay) print(fSync Task {name} completed) # 测试异步执行 async def async_task(name, delay): print(fAsync Task {name} started) await asyncio.sleep(delay) print(fAsync Task {name} completed) # 同步测试 def test_sync(): start time.time() sync_task(A, 2) sync_task(B, 1) sync_task(C, 3) end time.time() print(fSync execution time: {end - start:.2f} seconds) # 异步测试 async def test_async(): start time.time() await asyncio.gather( async_task(A, 2), async_task(B, 1), async_task(C, 3) ) end time.time() print(fAsync execution time: {end - start:.2f} seconds) if __name__ __main__: print(\nTesting synchronous execution:) test_sync() print(\nTesting asynchronous execution:) asyncio.run(test_async())高级应用场景1. 数据流处理def process_data(data): 处理数据流 for item in data: # 处理每个数据项 processed item * 2 yield processed # 使用生成器处理大型数据集 large_data range(1000000) processed_data process_data(large_data) # 按需获取处理结果 for i, value in enumerate(processed_data): if i 10: # 只处理前11个数据 break print(value)2. 异步网络请求import asyncio import aiohttp async def fetch_url(url): 异步获取URL内容 async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.text() async def main(): urls [ https://api.github.com, https://api.python.org, https://api.openai.com ] # 并发获取多个URL tasks [fetch_url(url) for url in urls] results await asyncio.gather(*tasks) for url, content in zip(urls, results): print(fURL: {url}, Content length: {len(content)}) asyncio.run(main())3. 异步文件操作import asyncio import aiofiles async def read_file(filename): 异步读取文件 async with aiofiles.open(filename, r) as f: content await f.read() return content async def write_file(filename, content): 异步写入文件 async with aiofiles.open(filename, w) as f: await f.write(content) async def main(): # 读取文件 content await read_file(input.txt) print(fRead content length: {len(content)}) # 处理内容 processed_content content.upper() # 写入文件 await write_file(output.txt, processed_content) print(File written successfully) asyncio.run(main())最佳实践1. 生成器最佳实践使用生成器表达式对于简单的转换使用生成器表达式比列表推导式更节省内存实现__iter__方法在类中实现__iter__方法返回生成器使类可迭代使用yield from在生成器中使用yield from委托给其他生成器2. 协程最佳实践使用async with对于需要资源管理的操作使用async with确保资源正确释放使用async for对于异步迭代器使用async for进行遍历避免阻塞操作在协程中避免使用阻塞操作如time.sleep()应使用asyncio.sleep()合理使用asyncio.gather()对于并发任务使用asyncio.gather()提高效率代码优化建议1. 生成器优化# 原始代码 def process_large_data(data): result [] for item in data: processed complex_calculation(item) result.append(processed) return result # 优化后代码 def process_large_data(data): for item in data: processed complex_calculation(item) yield processed2. 协程优化# 原始代码 async def fetch_multiple_urls(urls): results [] for url in urls: result await fetch_url(url) results.append(result) return results # 优化后代码 async def fetch_multiple_urls(urls): tasks [fetch_url(url) for url in urls] results await asyncio.gather(*tasks) return results3. 混合使用生成器和协程async def process_data_stream(data_generator): 处理数据生成器中的数据 tasks [] async with aiohttp.ClientSession() as session: for item in data_generator: # 为每个数据项创建异步任务 task process_item(session, item) tasks.append(task) # 每10个任务执行一次 if len(tasks) 10: await asyncio.gather(*tasks) tasks [] # 处理剩余任务 if tasks: await asyncio.gather(*tasks) async def process_item(session, item): 处理单个数据项 # 执行异步操作如网络请求 async with session.get(fhttps://api.example.com/{item}) as response: return await response.json()输入输出示例输入输出示例示例1生成器应用输入def prime_numbers(): 生成质数 num 2 while True: if all(num % i ! 0 for i in range(2, int(num**0.5) 1)): yield num num 1 # 使用生成器获取前10个质数 primes prime_numbers() for i in range(10): print(next(primes))输出2 3 5 7 11 13 17 19 23 29示例2协程应用输入import asyncio async def countdown(n): while n 0: print(fCountdown: {n}) await asyncio.sleep(1) n - 1 print(Go!) async def main(): # 并发执行两个倒计时 await asyncio.gather( countdown(3), countdown(5) ) asyncio.run(main())输出Countdown: 3 Countdown: 5 Countdown: 2 Countdown: 4 Countdown: 1 Countdown: 3 Go! Countdown: 2 Countdown: 1 Go!总结Python 生成器和协程是实现高效异步编程的强大工具。生成器通过按需生成值来节省内存而协程通过非阻塞 I/O 操作来提高程序的并发性能。核心优势特性优势适用场景生成器内存高效、代码简洁、惰性计算处理大型数据集、无限序列协程非阻塞 I/O、高并发、代码清晰网络请求、文件操作、数据库查询实际应用建议数据处理使用生成器处理大型数据集避免一次性加载所有数据到内存网络请求使用协程并发处理多个网络请求提高吞吐量文件操作使用异步文件操作提高 I/O 密集型任务的性能混合使用在复杂场景中结合生成器和协程充分发挥两者的优势通过合理使用生成器和协程我们可以编写更加高效、简洁、可维护的 Python 代码特别是在处理 I/O 密集型任务和大型数据集时它们能够显著提升程序的性能和可靠性。