关注+星标，每天学习Python新技能
来源：网络
作者：Python进阶者
关键词：Python生成器, 协程, yield, async/await, 异步编程, 并发
开头引言：
大家好，我是Python进阶者。今天我们要探索Python异步编程的演进之路——从简单的生成器到强大的协程，再到现代的async/await语法。这是一段从yield到async/await的技术进化史，也是Python成为高效异步编程语言的关键转折点。无论你是想理解异步编程的原理，还是希望写出更高效的程序，这篇文章都将为你揭开其中的奥秘。
一、生成器基础：yield的魔力
1.1 什么是生成器？
生成器是一种特殊的迭代器，它使用yield语句来惰性生成值，而不是一次性计算所有值。
defsimple_generator():
"""简单的生成器示例"""
print("开始执行")
yield1
print("继续执行")
yield2
print("结束执行")
yield3
# 使用生成器
gen = simple_generator()
print("第一次调用:")
print(next(gen)) # 输出: 开始执行 → 1
print("第二次调用:")
print(next(gen)) # 输出: 继续执行 → 2
print("第三次调用:")
print(next(gen)) # 输出: 结束执行 → 3
# print(next(gen)) # 会抛出 StopIteration
1.2 生成器的优势
内存效率对比：
import sys
# 列表方式（一次性加载所有数据）
defget_numbers_list(n):
numbers = []
for i inrange(n):
numbers.append(i)
return numbers
# 生成器方式（惰性生成）
defget_numbers_gen(n):
for i inrange(n):
yield i
# 测试内存使用
n = 1000000
list_memory = sys.getsizeof(get_numbers_list(n))
gen_memory = sys.getsizeof(get_numbers_gen(n))
print(f"列表内存占用: {list_memory / 1024 / 1024:.2f} MB")
print(f"生成器内存占用: {gen_memory / 1024 / 1024:.2f} MB")
输出结果：
列表内存占用: 8.00 MB
生成器内存占用: 0.00 MB
1.3 生成器表达式
# 列表推导式（立即计算）
squares_list = [x * x for x inrange(1000000)]
# 生成器表达式（惰性计算）
squares_gen = (x * x for x inrange(1000000))
print(f"列表大小: {sys.getsizeof(squares_list)} bytes")
print(f"生成器大小: {sys.getsizeof(squares_gen)} bytes")
二、生成器进阶：yield的双向通信
2.1 yield作为表达式
Python 2.5引入了yield表达式，允许生成器接收外部数据：
definteractive_generator():
"""支持双向通信的生成器"""
print("生成器启动")
received = yield"请发送数据"
print(f"收到: {received}")
received = yield"请发送更多数据"
print(f"收到: {received}")
yield"完成"
# 使用示例
gen = interactive_generator()
print(next(gen)) # 启动生成器，输出: "请发送数据"
print(gen.send("你好")) # 发送数据，输出: "收到: 你好" → "请发送更多数据"
print(gen.send("世界")) # 发送数据，输出: "收到: 世界" → "完成"
2.2 生成器作为协程
通过send()、throw()和close()方法，生成器可以当作简单的协程使用：
defcoroutine_example():
"""简单的协程示例"""
try:
whileTrue:
value = yield
print(f"处理值: {value}")
except GeneratorExit:
print("协程关闭")
except Exception as e:
print(f"发生异常: {e}")
# 使用示例
coro = coroutine_example()
next(coro) # 启动协程
coro.send(10) # 输出: 处理值: 10
coro.send(20) # 输出: 处理值: 20
coro.throw(ValueError("测试异常")) # 输出: 发生异常: 测试异常
coro.close() # 输出: 协程关闭
三、yield from：生成器委托
Python 3.3引入了yield from语法，简化了生成器的嵌套和委托：
3.1 基本用法
defsub_generator():
yieldfrom [1, 2, 3]
yieldfrom (4, 5, 6)
yieldfromrange(7, 10)
defmain_generator():
yieldfrom sub_generator()
yield"完成"
# 使用示例
for value in main_generator():
print(value, end=" ") # 输出: 1 2 3 4 5 6 7 8 9 完成
3.2 委托式协程
defdelegating_coroutine():
"""委托式协程示例"""
result = yieldfrom sub_coroutine()
print(f"子协程返回: {result}")
yield"主协程完成"
defsub_coroutine():
values = []
try:
whileTrue:
value = yield
values.append(value)
print(f"子协程收到: {value}")
except GeneratorExit:
returnsum(values)
# 使用示例
main = delegating_coroutine()
next(main) # 启动主协程
main.send(10) # 子协程收到: 10
main.send(20) # 子协程收到: 20
main.send(30) # 子协程收到: 30
try:
main.close() # 关闭协程
except StopIteration as e:
print(f"最终结果: {e.value}") # 输出: 子协程返回: 60 → 主协程完成
四、asyncio与原生协程
Python 3.4引入了asyncio模块，3.5引入了async/await语法，带来了真正的原生协程。
4.1 异步编程的基本概念
import asyncio
import time
asyncdefsay_after(delay, message):
"""异步函数示例"""
await asyncio.sleep(delay)
print(message)
returnf"完成: {message}"
asyncdefmain():
"""主协程"""
print(f"开始时间: {time.strftime('%X')}")
# 顺序执行
result1 = await say_after(2, "你好")
result2 = await say_after(1, "世界")
print(f"结束时间: {time.strftime('%X')}")
print(f"结果: {result1}, {result2}")
# 运行协程
asyncio.run(main())
4.2 并发执行任务
asyncdefconcurrent_main():
"""并发执行示例"""
print(f"开始时间: {time.strftime('%X')}")
# 创建任务并发执行
task1 = asyncio.create_task(say_after(2, "你好"))
task2 = asyncio.create_task(say_after(1, "世界"))
# 等待所有任务完成
results = await asyncio.gather(task1, task2)
print(f"结束时间: {time.strftime('%X')}")
print(f"所有结果: {results}")
asyncio.run(concurrent_main())
五、async/await语法详解
5.1 async函数声明
# 传统生成器协程
defold_coroutine():
value = yield
return value
# 现代异步函数
asyncdefmodern_coroutine():
# 不能再使用yield
# 必须使用await调用其他异步函数
result = await some_async_operation()
return result
5.2 await表达式
await只能在async函数中使用，用于等待可等待对象：
asyncdefexample():
# 等待协程
result1 = await async_function()
# 等待Future对象
result2 = await asyncio.Future()
# 等待任务
task = asyncio.create_task(async_function())
result3 = await task
return result1, result2, result3
六、实际应用案例
6.1 异步网络请求
import aiohttp
import asyncio
asyncdeffetch_url(session, url):
"""异步获取URL内容"""
try:
asyncwith session.get(url, timeout=10) as response:
content = await response.text()
returnf"{url}: {len(content)} bytes"
except Exception as e:
returnf"{url}: 错误 - {e}"
asyncdefmain():
"""主函数"""
urls = [
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/2",
"https://httpbin.org/delay/3",
"https://invalid-url-test.com"
]
asyncwith aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
# 运行
asyncio.run(main())
6.2 异步文件操作
import aiofiles
import asyncio
asyncdefasync_file_operations():
"""异步文件操作示例"""
# 异步写入文件
asyncwith aiofiles.open('test.txt', 'w') as f:
await f.write('Hello, async world!\n')
await f.write('This is async file IO.\n')
# 异步读取文件
asyncwith aiofiles.open('test.txt', 'r') as f:
content = await f.read()
print("文件内容:")
print(content)
# 异步逐行读取
asyncwith aiofiles.open('test.txt', 'r') as f:
asyncfor line in f:
print(f"行: {line.strip()}")
asyncio.run(async_file_operations())
6.3 生产者-消费者模式
import asyncio
import random
asyncdefproducer(queue, producer_id):
"""生产者协程"""
for i inrange(5):
item = f"产品-{producer_id}-{i}"
await asyncio.sleep(random.uniform(0.1, 0.5))
await queue.put(item)
print(f"生产者 {producer_id} 生产了: {item}")
await queue.put(None) # 结束信号
asyncdefconsumer(queue, consumer_id):
"""消费者协程"""
whileTrue:
item = await queue.get()
if item isNone:
await queue.put(None) # 传递给其他消费者
break
print(f"消费者 {consumer_id} 消费了: {item}")
await asyncio.sleep(random.uniform(0.2, 0.8))
asyncdefmain():
"""主函数"""
queue = asyncio.Queue(maxsize=10)
# 创建生产者和消费者
producers = [producer(queue, i) for i inrange(3)]
consumers = [consumer(queue, i) for i inrange(2)]
# 运行所有协程
await asyncio.gather(*producers, *consumers)
asyncio.run(main())
七、性能对比：同步 vs 异步
7.1 I/O密集型任务对比
import asyncio
import time
import requests
import aiohttp
defsync_fetch(urls):
"""同步获取多个URL"""
results = []
for url in urls:
response = requests.get(url)
results.append(f"{url}: {len(response.text)} bytes")
return results
asyncdefasync_fetch(urls):
"""异步获取多个URL"""
asyncwith aiohttp.ClientSession() as session:
tasks = []
for url in urls:
task = asyncio.create_task(
fetch_url(session, url)
)
tasks.append(task)
returnawait asyncio.gather(*tasks)
asyncdeffetch_url(session, url):
"""辅助函数"""
asyncwith session.get(url) as response:
content = await response.text()
returnf"{url}: {len(content)} bytes"
# 性能测试
urls = ["https://httpbin.org/delay/1"] * 10
print("同步版本:")
start = time.time()
sync_fetch(urls)
sync_time = time.time() - start
print(f"同步耗时: {sync_time:.2f}秒")
print("\n异步版本:")
start = time.time()
asyncio.run(async_fetch(urls))
async_time = time.time() - start
print(f"异步耗时: {async_time:.2f}秒")
print(f"\n性能提升: {sync_time/async_time:.1f}倍")
八、错误处理与调试
8.1 异步异常处理
asyncdefrisky_operation():
"""可能失败的异步操作"""
await asyncio.sleep(0.1)
if random.random() < 0.3:
raise ValueError("随机失败")
return"成功"
asyncdefrobust_main():
"""健壮的异步错误处理"""
tasks = []
for i inrange(10):
task = asyncio.create_task(risky_operation())
tasks.append(task)
results = []
for task in tasks:
try:
result = await task
results.append(result)
except ValueError as e:
print(f"任务失败: {e}")
results.append("失败")
print(f"完成结果: {results}")
asyncio.run(robust_main())
8.2 异步调试技巧
import logging
logging.basicConfig(level=logging.DEBUG)
asyncdefdebug_coroutine():
"""可调试的协程"""
logging.debug("协程开始")
await asyncio.sleep(0.1)
logging.debug("第一步完成")
await asyncio.sleep(0.1)
logging.debug("第二步完成")
return"完成"
# 设置调试模式
asyncio.run(debug_coroutine(), debug=True)
九、最佳实践与常见陷阱
9.1 最佳实践
- 1.
使用async/await而不是旧式协程
- 2.
合理控制并发数量
- 3.
正确处理异常
- 4.
使用适当的超时设置
- 5.
避免在异步代码中阻塞
9.2 常见陷阱
# 错误示例：在异步函数中阻塞
asyncdefbad_example():
# 这会阻塞事件循环！
time.sleep(1) # 错误！
# 应该使用: await asyncio.sleep(1)
# 错误示例：忘记await
asyncdefanother_bad_example():
result = some_async_function() # 错误！忘记await
# 应该使用: result = await some_async_function()
# 正确示例
asyncdefgood_example():
await asyncio.sleep(1) # 正确
result = await some_async_function() # 正确
十、总结：技术演进路线
版本 |
特性 |
意义 |
Python 2.2 |
生成器（yield） |
引入了惰性计算 |
Python 2.5 |
yield表达式 |
支持双向通信 |
Python 3.3 |
yield from |
生成器委托 |
Python 3.4 |
asyncio模块 |
异步编程框架 |
Python 3.5 |
async/await |
原生协程支持 |
Python 3.7 |
asyncio.run() |
简化异步入口 |
选择建议：
- •
✅ I/O密集型任务：使用async/await
- •
✅ CPU密集型任务：使用多进程+asyncio
- •
✅ 简单惰性计算：使用生成器
- •
✅ 复杂异步逻辑：使用asyncio生态系统
扫描下方二维码，咨询七七老师！
免费获取Python公开课和大佬打包整理的几百G的学习资料，内容包含但不限于Python电子书，学习路线，职业规划，人工智能、游戏开发、网站开发、自动化 、数据抓取教程，面试题，副业接单，Python高薪就业等等~
学习遇到瓶颈，缺乏学习方向、缺乏实战、就业问题、副业问题、百度搜索1个问题要几个小时解答、安装软件需要半个小时左右、不会英语不知道怎么学、不会汉化、想往这个方面去发展的、做数据分析提高效率、做全栈开发、小程序、游戏开发、脚本开发、财务分析、股票交易分析、网站开发、小程序员、自动化处理、人工智能、机器学习，可以咨询七七老师
推荐阅读
基于Django+Celery+Acunetix的漏洞扫描器源代码+数据库安装使用说明，实现了漏洞扫描、端口扫描和后台扫描等功能
python课程设计基于Django + vue实现的运动商城系统源代码+数据库+详细项目文档
告别混乱：一文掌握Python虚拟环境最佳实践
装饰器进阶：带参数的装饰器和类装饰器
点击 阅读原文 了解更多
