当前位置：首页 > news >正文

Python异步编程深度解析

news 2026/5/7 6:28:42

Python异步编程深度解析：从原理到实战

深入理解asyncio事件循环、协程机制，掌握高并发编程的核心技术

前言

在开发一个高并发的数据采集系统时，我发现传统的同步代码在处理大量IO操作时性能极差。通过引入异步编程，系统的并发能力提升了数十倍。本文将从实际问题出发，深入讲解Python异步编程的原理和实战技巧。

本文特点：

从实际问题出发，不是为了讲技术而讲技术
包含大量可运行的代码示例
分享真实的踩坑经验和解决方案
对比不同方案的性能差异

一、为什么需要异步编程

1.1 同步编程的问题

看一个典型的场景：爬取100个网页

importrequestsimporttimedeffetch_url(url:str)->str:"""同步获取URL内容"""response=requests.get(url,timeout=10)returnresponse.textdeffetch_all_sync(urls:list)->list:"""同步爬取所有URL"""results=[]forurlinurls:try:content=fetch_url(url)results.append(content)exceptExceptionase:results.append(None)returnresults# 测试同步性能urls=["https://httpbin.org/delay/1"]*10# 每个请求延迟1秒start=time.time()results=fetch_all_sync(urls)print(f"同步耗时:{time.time()-start:.2f}秒")# 约10秒

问题分析：

每个请求需要等待1秒
10个请求串行执行，总耗时约10秒
CPU大部分时间在等待网络IO，浪费严重

1.2 异步编程的优势

importaiohttpimportasyncioimporttimeasyncdeffetch_url_async(session:aiohttp.ClientSession,url:str)->str:"""异步获取URL内容"""asyncwithsession.get(url,timeout=aiohttp.ClientTimeout(total=10))asresponse:returnawaitresponse.text()asyncdeffetch_all_async(urls:list)->list:"""异步爬取所有URL"""asyncwithaiohttp.ClientSession()assession:tasks=[fetch_url_async(session,url)forurlinurls]returnawaitasyncio.gather(*tasks,return_exceptions=True)# 测试异步性能urls=["https://httpbin.org/delay/1"]*10start=time.time()results=asyncio.run(fetch_all_async(urls))print(f"异步耗时:{time.time()-start:.2f}秒")# 约1秒

性能对比：

同步：10个请求，每个1秒，总耗时约10秒
异步：10个请求并发执行，总耗时约1秒
性能提升：10倍

二、asyncio核心概念

2.1 事件循环（Event Loop）

事件循环是asyncio的核心，它负责调度所有的协程任务。

importasyncio# 获取事件循环loop=asyncio.get_event_loop()# 或者创建新的事件循环loop=asyncio.new_event_loop()asyncio.set_event_loop(loop)

事件循环的工作原理：

┌─────────────────────────────────────────┐ │ 事件循环 │ │ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ 协程任务1 │ │ 协程任务2 │ │ 协程任务3 │ │ │ └─────────┘ └─────────┘ └─────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────────────────┐│ │ │ 任务调度器 ││ │ └─────────────────────────────────────┘│ │ │ │ │ ▼ │ │ ┌─────────────────────────────────────┐│ │ │ IO多路复用器 ││ │ │ (select/epoll/kqueue) ││ │ └─────────────────────────────────────┘│ │ │ │ │ ▼ │ │ ┌─────────────────────────────────────┐│ │ │ 操作系统内核 ││ │ └─────────────────────────────────────┘│ └─────────────────────────────────────────┘

2.2 协程（Coroutine）

协程是异步编程的基本单位，使用async def定义。

importasyncioasyncdefsimple_coroutine():"""一个简单的协程"""print("协程开始")awaitasyncio.sleep(1)# 模拟IO操作print("协程结束")return"结果"# 运行协程result=asyncio.run(simple_coroutine())print(result)

协程vs函数：

# 普通函数defnormal_function():return42# 协程函数asyncdefcoroutine_function():return42# 调用普通函数直接返回结果result=normal_function()# result = 42# 调用协程函数返回协程对象coro=coroutine_function()# coro = <coroutine object>result=asyncio.run(coro)# result = 42

2.3 await关键字

await用于等待一个异步操作完成。

importasyncioasyncdeffetch_data():"""模拟获取数据"""print("开始获取数据...")awaitasyncio.sleep(2)# 模拟网络请求print("数据获取完成")return{"name":"张三","age":25}asyncdefprocess_data():"""处理数据"""# await会暂停当前协程，等待fetch_data完成data=awaitfetch_data()print(f"处理数据:{data}")returnf"处理结果:{data['name']}"asyncio.run(process_data())

await的原理：

# await做了什么？awaitsome_coroutine()# 等价于：# 1. 暂停当前协程# 2. 将some_coroutine加入事件循环# 3. 等待some_coroutine完成# 4. 恢复当前协程，获取结果

三、异步编程实战

3.1 异步HTTP请求

importaiohttpimportasynciofromtypingimportList,DictclassAsyncHTTPClient:"""异步HTTP客户端"""def__init__(self,max_concurrent:int=10):self.semaphore=asyncio.Semaphore(max_concurrent)self.session=Noneasyncdef__aenter__(self):self.session=aiohttp.ClientSession()returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):ifself.session:awaitself.session.close()asyncdefget(self,url:str,**kwargs)->Dict:"""发送GET请求"""asyncwithself.semaphore:# 限制并发数try:asyncwithself.session.get(url,**kwargs)asresponse:return{"status":response.status,"data":awaitresponse.json(),"url":url}exceptExceptionase:return{"status":-1,"error":str(e),"url":url}asyncdefget_multiple(self,urls:List[str])->List[Dict]:"""并发获取多个URL"""tasks=[self.get(url)forurlinurls]returnawaitasyncio.gather(*tasks)# 使用示例asyncdefmain():urls=["https://httpbin.org/get","https://httpbin.org/delay/1","https://httpbin.org/delay/2",]asyncwithAsyncHTTPClient(max_concurrent=5)asclient:results=awaitclient.get_multiple(urls)forresultinresults:print(f"URL:{result['url']}")print(f"Status:{result['status']}