企业官网建设流程全解析

Python 异步编程实战指南（asyncio）

Python 的 GIL 让多线程写 CPU 密集任务很尴尬，但 IO 密集场景异步才是王道——asyncio 是 Python 异步编程的核心，也是 CSDN Python 板块的热门话题。本文从并发模型讲起，覆盖协程、事件循环、async/await、异步库、asyncio 实战、与多线程对比、常见坑。

一、为什么需要异步

1.1 Python 并发困境

多线程：GIL（全局解释器锁）→ 同一时刻只有一个线程执行字节码 → CPU 密集任务多线程不加速 → IO 密集任务：线程切换开销大，万级连接就崩

结论：Python 处理高并发 IO 的正确姿势是异步（单线程 + 事件循环），不是多线程。

1.2 三种并发模型对比

经验：

• IO 密集 + 高并发 →asyncio；
• CPU 密集 → 多进程（multiprocessing）；
• 简单脚本 → 同步就行，别过度设计。

二、协程与 async/await

2.1 协程是什么

协程（Coroutine）：可以暂停和恢复的函数——遇到 IO 主动让出控制权，IO 完成再回来。

同步：请求1 → 等1秒 → 请求2 → 等1秒 → 总耗时 2 秒 异步：请求1发出 → 等的同时发出请求2 → 两个一起回来 → 总耗时 1 秒

2.2 基本语法

importasyncio# 定义协程函数（async def）asyncdeffetch_data():print("开始请求")awaitasyncio.sleep(1)# 模拟 IO（await 让出控制权）print("请求完成")return"数据"# 运行asyncdefmain():result=awaitfetch_data()print(result)asyncio.run(main())

关键：

• async def：定义协程函数；
• await：等待另一个协程（遇到它就让出控制权）；
• asyncio.run()：启动事件循环。

注意：协程函数调用时不执行——必须 await 或放进事件循环。

三、事件循环

3.1 概念

事件循环（Event Loop）：单线程调度器——轮询就绪任务并执行。

事件循环： 1. 取出一个就绪协程执行 2. 遇到 await（IO）→ 挂起，记录回调 3. 继续执行其他就绪协程 4. IO 完成 → 恢复挂起的协程 5. 循环

3.2 任务（Task）与并发

importasyncioasyncdeftask(name,seconds):print(f"{name}开始")awaitasyncio.sleep(seconds)print(f"{name}完成")returnnameasyncdefmain():# 并发执行（不是 await 顺序执行！）t1=asyncio.create_task(task("任务A",2))t2=asyncio.create_task(task("任务B",1))# 等待全部完成results=awaitasyncio.gather(t1,t2)print("全部完成:",results)asyncio.run(main())# 输出顺序：# 任务A 开始# 任务B 开始# 任务B 完成（1秒后）# 任务A 完成（2秒后）# 总耗时 2 秒（而不是 3 秒）

对比：

# 错误：顺序 await（串行，总耗时 3 秒）asyncdefwrong():awaittask("A",2)# 等 A 完成awaittask("B",1)# 才等 B# 正确：create_task + gather（并发，总耗时 2 秒）asyncdefright():awaitasyncio.gather(task("A",2),task("B",1),)

这是异步编程最大的思维转变：把"先 A 后 B"改成"同时发起，一起等"。

四、常用 API

4.1 gather：并发聚合

# 等所有完成，拿到结果列表results=awaitasyncio.gather(fetch("https://a.com"),fetch("https://b.com"),fetch("https://c.com"),return_exceptions=True,# 单个失败不影响其他（不抛异常）)# 处理结果forrinresults:ifisinstance(r,Exception):print("失败:",r)else:print("成功:",r)

4.2 wait：控制等待方式

# 等任一完成即可（FIRST_COMPLETED）done,pending=awaitasyncio.wait([t1,t2,t3],return_when=asyncio.FIRST_COMPLETED,)# 限时等待done,pending=awaitasyncio.wait(tasks,timeout=5)

4.3 超时控制

# 方案1：wait_fortry:result=awaitasyncio.wait_for(slow_task(),timeout=3)exceptasyncio.TimeoutError:print("超时了")# 方案2：asyncio.timeout（3.11+）try:asyncwithasyncio.timeout(3):result=awaitslow_task()exceptTimeoutError:print("超时了")

4.4 协程间通信：Queue

# 生产者-消费者asyncdefproducer(queue):foriinrange(10):awaitqueue.put(i)print(f"生产:{i}")awaitasyncio.sleep(0.1)asyncdefconsumer(queue):whileTrue:item=awaitqueue.get()print(f"消费:{item}")queue.task_done()asyncdefmain():queue=asyncio.Queue(maxsize=5)# 一个生产者 + 两个消费者awaitasyncio.gather(producer(queue),consumer(queue),consumer(queue),)

五、异步 IO（aiohttp）

5.1 异步 HTTP 请求

pipinstallaiohttp

importaiohttpimportasyncioasyncdeffetch(session,url):try:asyncwithsession.get(url,timeout=10)asresp:returnawaitresp.json()# 异步读响应exceptExceptionase:return{"error":str(e),"url":url}asyncdefmain():urls=[f"https://api.example.com/page/{i}"foriinrange(50)]asyncwithaiohttp.ClientSession()assession:# 50 个请求并发（总耗时 ≈ 单个请求耗时，不是 50 倍）results=awaitasyncio.gather(*(fetch(session,url)forurlinurls))print(f"完成{len(results)}个请求")forrinresults[:3]:print(r)asyncio.run(main())

对比 requests（同步）：50 个请求串行 50 秒 vs 并发 2 秒——量级差距。

5.2 并发限制（信号量）

# 防止一次发太多（礼貌 + 防封）semaphore=asyncio.Semaphore(10)# 最多 10 个并发asyncdeffetch_limited(session,url):asyncwithsemaphore:asyncwithsession.get(url)asresp:returnawaitresp.json()asyncdefmain():asyncwithaiohttp.ClientSession()assession:tasks=[fetch_limited(session,u)foruinurls]results=awaitasyncio.gather(*tasks)

六、异步数据库

6.1 SQLAlchemy 异步

fromsqlalchemy.ext.asyncioimportcreate_async_engine,AsyncSessionfromsqlalchemyimporttext engine=create_async_engine("mysql+aiomysql://root:123456@localhost/mall")asyncdefquery():asyncwithAsyncSession(engine)assession:result=awaitsession.execute(text("SELECT * FROM product LIMIT 10"))rows=result.fetchall()returnrowsasyncdefmain():rows=awaitquery()forrinrows:print(r)asyncio.run(main())

6.2 同步库的坑

# 错误：在协程里用同步阻塞库（卡死整个事件循环！）asyncdefbad():time.sleep(1)# 阻塞 → 所有协程都等requests.get(url)# 阻塞 → 灾难# 正确1：改用异步库（aiomysql、aiohttp、asyncpg）# 正确2：实在没有异步库，丢线程池importasyncioasyncdefgood():# 阻塞调用丢到线程池（不阻塞事件循环）result=awaitasyncio.to_thread(requests.get,url)

记住：事件循环里禁止任何同步阻塞调用——这是异步代码性能杀手。

七、实战：并发抓取 + 处理

importaiohttpimportasyncioasyncdeffetch_product(session,product_id):url=f"https://api.example.com/products/{product_id}"asyncwithsession.get(url)asresp:ifresp.status==200:data=awaitresp.json()return{"id":product_id,"price":data["price"]}return{"id":product_id,"error":resp.status}asyncdefprocess(session,product_id):# 抓取 + 简单处理data=awaitfetch_product(session,product_id)if"price"indata:data["price"]=data["price"]*0.9# 打 9 折returndataasyncdefmain():ids=list(range(1,201))# 200 个商品asyncwithaiohttp.ClientSession()assession:# 限制并发 20，抓 200 个sem=asyncio.Semaphore(20)asyncdeflimited(pid):asyncwithsem:returnawaitprocess(session,pid)results=awaitasyncio.gather(*(limited(i)foriinids))success=[rforrinresultsif"price"inr]print(f"成功{len(success)}/{len(ids)}，平均价格: "f"{sum(r['price']forrinsuccess)/len(success):.2f}")asyncio.run(main())

八、asyncio 与多线程/多进程结合

8.1 CPU 密集任务

# asyncio 不擅长 CPU 计算——丢给进程池importasynciofromconcurrent.futuresimportProcessPoolExecutordefheavy_calc(n):# CPU 密集（同步函数）returnsum(i*iforiinrange(n))asyncdefmain():withProcessPoolExecutor(max_workers=4)aspool:# 不阻塞事件循环results=awaitasyncio.gather(*[asyncio.get_event_loop().run_in_executor(pool,heavy_calc,10**7)for_inrange(4)])print(results)asyncio.run(main())

8.2 与线程库共存

# FastAPI/Starlette 自动处理：async 路由走事件循环，同步路由丢线程池# （FastAPI 篇讲过——直接用就行）

九、常见坑速查

本章小结

Python 异步的核心心智：单线程事件循环 + await 让出控制权 + gather 并发聚合。写异步代码记住三件事：协程里禁止同步阻塞、并发用 create_task/gather 而不是顺序 await、海量请求加 Semaphore 限流。对爬虫、API 聚合、微服务网关这类 IO 密集场景，asyncio 能把性能提升一个数量级。

下一篇讲 CI/CD 流水线——现代软件交付的标配。

（全文完）