Python 异步编程
Python 异步编程是一种基于非阻塞 IO 模型的并发编程范式,核心目标是在处理 IO 密集型任务(如网络请求、文件读写、数据库交互)时,通过高效的任务调度减少等待时间,最大化 CPU 利用率。异步编程通过事件循环实现任务调度:当一个任务因 IO 操作需要等待时,事件循环会暂停该任务,切换到其他就绪任务;当 IO 操作完成(如响应到达),事件循环再恢复原任务的执行。
核心优势:
[*]单线程内实现并发,避免多线程的上下文切换开销。
[*]针对 IO 密集型任务,性能提升显著(通常是同步方式的数倍到数十倍)。
1、核心组件
1.1 事件循环
事件循环是异步编程的 “心脏”,负责任务调度、IO 事件监听、状态管理。
# 伪代码
任务列表 = [ 任务1,任务2,任务3,...]
while True:
可执行的任务列表,已完成的任务列表 = 去任务列表中检查所有的任务,将'可执行'和'已完成'的任务返回
for 就绪任务 in 已准备就绪的任务列表:
执行已就绪的任务
for 已完成的任务 in 已完成的任务列表:
在任务列表中移除 已完成的任务
如果 任务列表 中的任务都已完成,则终止循环关键细节:
[*]事件循环在单线程中运行,所有任务都在这个线程内切换执行。
[*]仅当任务遇到 await(表示需要等待 IO)时才会切换,纯计算任务不会触发切换。
1.2 协程
协程是异步任务的基本单元,是一种用户态的上下文切换技术,其实就是通过一个线程实现代码块相互切换执行,本质是可暂停 / 恢复的函数,通过 async def 定义。与普通函数的区别在于:
[*]调用协程不会立即执行,而是返回一个协程对象。
[*]必须通过事件循环调度(如 await、create_task)才能执行。
# 协程的定义与状态
import asyncio
async def my_coroutine():
print("协程开始")
await asyncio.sleep(1)# 暂停点:释放CPU,允许切换
print("协程结束")
return "结果"
# 协程对象(未执行)
coro = my_coroutine()
print(type(coro))# <class 'coroutine'>
# 必须通过事件循环执行
async def main():
result = await coro# 调度执行,等待结果
print(result)# 输出:结果
asyncio.run(main())协程的生命周期:
[*]创建:coro = my_coroutine() → 未执行状态。
[*]运行:await coro 或 create_task(coro) → 进入事件循环。
[*]暂停:执行到 await 语句 → 等待 IO 时挂起。
[*]恢复:IO 完成 → 从暂停点继续执行。
[*]完成:执行到函数末尾 → 返回结果或抛出异常。
1.3 任务
任务是协程的包装器,由事件循环直接调度,用于实现并发。任务会将协程注册到事件循环,并跟踪其状态(运行中 / 已完成 / 已取消)。
async def task_func(name, delay):
print(f"任务 name={name}, delay={delay} === 111")
await asyncio.sleep(delay)
print(f"任务 name={name}, delay={delay} === 222")
return f"任务 {name} 完成"
async def main():
# 创建任务(立即加入事件循环,开始调度)
task1 = asyncio.create_task(task_func("A", 1))
task2 = asyncio.create_task(task_func("B", 2))
print("任务状态:", task1.done())# False(未完成)
# 等待任务完成并获取结果
result1 = await task1
result2 = await task2
print("结果:", result1, result2)# 任务 A 完成 任务 B 完成
print("任务状态:", task1.done())# True(已完成)
asyncio.run(main())任务的核心方法:
[*]task.done():判断任务是否完成。
[*]task.result():获取任务返回值(任务未完成时调用会报错)。
[*]task.cancel():取消任务(触发 CancelledError)。
[*]task.add_done_callback(func):注册回调函数(任务完成后执行)。
更常用写法:
async def task_func(name, delay):
print(f"任务 name={name}, delay={delay} === 111")
await asyncio.sleep(delay)
print(f"任务 name={name}, delay={delay} === 222")
return f"任务 {name} 完成"
async def main():
# 创建任务(立即加入事件循环,开始调度)
task_list = [
asyncio.create_task(task_func("A", 1), name="task_A"),
asyncio.create_task(task_func("B", 2), name="task_B"),
]
done, pending = await asyncio.wait(task_list, timeout=None)
# 等待任务完成并获取结果
print(done)
asyncio.run(main())1.4 Future 对象
Future 是异步操作结果的容器,表示 “未来可能完成的操作”。任务(Task)是 Future 的子类,因此具备 Future 的所有特性,task对象内部await结果的处理是基于future的:
[*]存储异步操作的状态(PENDING/FINISHED/CANCELLED)。
[*]提供结果设置(set_result())和异常设置(set_exception())方法。
[*]支持通过 await 获取结果,或注册回调函数。
async def main():
# 创建一个空的Future对象
future = asyncio.Future()
# 定义一个设置Future结果的协程
async def set_future_result():
await asyncio.sleep(1)
future.set_result("Future 结果")# 设置结果,标记为完成
# 并发执行:设置结果的协程 + 等待结果的操作
asyncio.create_task(set_future_result())
result = await future# 等待Future完成
print(result)# 输出:Future 结果
asyncio.run(main())Task 与 Future 的关系:
[*]Task 继承自 Future,是 “可执行的 Future”(绑定了协程)。
[*]Future 更底层,通常用于手动管理异步操作结果(如包装回调式 API)。
2、基础语法和核心API
2.1 async/await 语法
async/await 是 Python 3.5+ 引入的异步语法糖,用于定义协程和暂停执行:
[*]async def:定义协程函数(返回协程对象)。
[*]await:暂停协程,等待另一个协程 / Future/Task 完成,只能在协程内部使用。
async def nested():
return 42
async def main():
# 直接调用协程不会执行,必须用await
result = await nested()# 等待nested完成,获取结果
print(result)# 42
asyncio.run(main())注意:
[*]普通函数中不能使用 await(会报 SyntaxError)。
[*]await 后面必须是 “可等待对象”(协程、Task、Future)
2.2 事件循环的启动与管理
Python 3.7+ 推荐用 asyncio.run() 启动事件循环(自动创建、运行、关闭循环),低版本需手动管理:
# Python 3.7+ 推荐方式
async def main():
await asyncio.sleep(1)
print("完成")
asyncio.run(main())# 自动处理事件循环的生命周期
# 低版本手动管理方式(3.6及以下)
loop = asyncio.get_event_loop()# 获取事件循环
try:
loop.run_until_complete(main())# 运行直到协程完成
finally:
loop.close()# 关闭循环2.3 并发任务管理
2.3.1 asyncio.gather()
批量并发与结果聚合
gather() 用于同时运行多个可等待对象,按输入顺序返回结果,适合需要统一收集结果的场景。
async def task(i):
await asyncio.sleep(i)
return i
async def main():
# 并发执行3个任务
results = await asyncio.gather(
task(1),
task(2),
task(0.5)
)
print(results)# (按输入顺序,而非完成顺序)
asyncio.run(main())高级参数:
[*]return_exceptions=True:将异常作为结果返回,不中断其他任务。
async def faulty_task():
raise ValueError("出错了")
async def main():
results = await asyncio.gather(
faulty_task(),
task(1),
return_exceptions=True# 异常会被包装到结果中
)
print(results)#
2.3.2 asyncio.wait()
灵活控制任务完成条件
wait() 比 gather() 更灵活,支持按 “第一个完成”“所有完成” 等条件返回,返回值是已完成和未完成的任务集合。
async def main():
tasks =
# 等待所有任务完成(默认)
done, pending = await asyncio.wait(tasks)
print("已完成任务数:", len(done))# 3
print("未完成任务数:", len(pending))# 0
# 等待第一个任务完成
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
print("第一个完成的任务结果:", )# return_when 可选值:
[*]FIRST_COMPLETED:第一个任务完成时返回。
[*]FIRST_EXCEPTION:第一个任务抛出异常时返回(无异常则等所有完成)。
[*]ALL_COMPLETED:所有任务完成时返回(默认)。
3、同步代码的异步化:兼容旧库
实际开发中常需在异步程序中调用同步阻塞库(如 requests、pymysql),直接调用会阻塞事件循环,需通过线程池异步执行。
3.1 核心方法:loop.run_in_executor()
该方法将同步函数提交到线程池执行,返回 Future 对象,可通过 await 获取结果。
import asyncio
import requests# 同步阻塞库
# 同步函数(阻塞)
def sync_get(url):
return requests.get(url).status_code
async def async_get(url):
# 获取事件循环
loop = asyncio.get_event_loop()
# 提交到线程池执行(None 表示使用默认线程池)
future = loop.run_in_executor(
None,# 线程池执行器(可选自定义)
sync_get,# 同步函数
url# 函数参数
)
return await future# 等待线程池结果
async def main():
urls = ["https://www.baidu.com", "https://www.github.com"]
# 并发执行同步函数的异步包装
results = await asyncio.gather(*)
print("结果:", results)#
asyncio.run(main())3.2 自定义线程池
默认线程池大小有限(通常为 CPU 核心数 * 5),高并发场景可自定义线程池:
from concurrent.futures import ThreadPoolExecutor
async def main():
# 自定义线程池(最大10个线程)
executor = ThreadPoolExecutor(max_workers=10)
loop = asyncio.get_event_loop()
# 使用自定义线程池
future = loop.run_in_executor(executor, sync_get, "https://www.baidu.com")
print(await future)# 2004、异步 IO 实战
4.1异步网络请求(aiohttp)
aiohttp 是异步 HTTP 客户端 / 服务器库,支持异步请求、连接池、超时控制等,是替代同步 requests 的最佳选择。
并发爬取网页(带超时与重试)
import asyncio
import aiohttp
from aiohttp import ClientTimeout
async def fetch(session, url, retry=3):
"""带重试机制的异步请求"""
timeout = ClientTimeout(total=10)# 超时控制(10秒)
try:
async with session.get(url, timeout=timeout) as response:
return {
"url": url,
"status": response.status,
"length": len(await response.text())
}
except Exception as e:
if retry > 0:
print(f"请求 {url} 失败,重试 {retry-1} 次: {e}")
await asyncio.sleep(1)# 重试前等待1秒
return await fetch(session, url, retry-1)
return {"url": url, "error": str(e)}
async def main():
urls = [
"https://www.baidu.com",
"https://www.github.com",
"https://www.python.org",
"https://invalid.url"
]
# 创建会话(复用连接,提高效率)
async with aiohttp.ClientSession() as session:
# 生成任务列表
tasks =
# 并发执行
results = await asyncio.gather(*tasks)
# 输出结果
for res in results:
if "error" in res:
print(f"{res['url']}: {res['error']}")
else:
print(f"{res['url']} | 状态: {res['status']} | 长度: {res['length']}")
asyncio.run(main())4.2 异步文件操作(aiofiles)
传统 open() 是同步阻塞的,aiofiles 提供异步文件读写,支持 async with 和 await 语法。
异步读写多文件
import asyncio
import aiofiles
async def write_file(filename, content):
"""异步写入文件"""
async with aiofiles.open(filename, 'w', encoding='utf-8') as f:
await f.write(content)# 异步写入
print(f"已写入: {filename}")
async def read_file(filename):
"""异步读取文件"""
async with aiofiles.open(filename, 'r', encoding='utf-8') as f:
content = await f.read()# 异步读取
return filename, content
async def main():
# 并发写入3个文件
await asyncio.gather(
write_file("file1.txt", "异步文件1"),
write_file("file2.txt", "异步文件2"),
write_file("file3.txt", "异步文件3")
)
# 并发读取文件
files = ["file1.txt", "file2.txt", "file3.txt"]
results = await asyncio.gather(*)
# 打印内容
for name, content in results:
print(f"{name} 内容: {content}")
asyncio.run(main())4.3 异步数据库操作(aiomysql)
aiomysql 是 MySQL 的异步驱动,支持异步连接、查询、事务,避免同步 pymysql 的阻塞问题。
异步查询 MySQL
import asyncio
import aiomysql
async def query_db():
# 建立异步连接
connection = await aiomysql.connect(
host='localhost',
port=3306,
user='root',
password='password',
db='test',
autocommit=True
)
try:
# 创建游标
async with connection.cursor(aiomysql.DictCursor) as cursor:
# 异步执行查询
await cursor.execute("SELECT * FROM users LIMIT 3")
# 异步获取结果
results = await cursor.fetchall()
print("查询结果:", results)
finally:
# 关闭连接
connection.close()
asyncio.run(query_db())5、总结
Python 异步编程通过事件循环驱动的任务切换,实现了 IO 密集型任务的高效并发。核心组件包括协程(任务单元)、事件循环(调度中心)、任务(并发单元)和 Future(结果容器)。
来源:程序园用户自行投稿发布,如果侵权,请联系站长删除
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作! 新版吗?好像是停更了吧。
页:
[1]