Python异步编程终极指南：用协程与事件循环重构你的高并发系统-牛翰网

title: Python异步编程终极指南：用协程与事件循环重构你的高并发系统
date: 2025/2/24
updated: 2025/2/24
author: cmdragon

excerpt:
深入剖析Python异步编程的核心机制。你将掌握：\n 事件循环的底层实现原理与调度算法\n async/await协程的6种高级用法模式\n 异步HTTP请求的性能优化技巧（速度提升15倍+）\n 常见异步陷阱的26种解决方案

categories:

后端开发
FastAPI

tags:

Python异步革命
asyncio黑科技
协程深度解析
事件循环架构
高性能HTTP
并发编程
异步调试

图片[1]-Python异步编程终极指南：用协程与事件循环重构你的高并发系统-牛翰网

扫描二维码关注或者微信搜一搜：编程智域前端至全栈交流与成长

摘要

深入剖析Python异步编程的核心机制。你将掌握：

事件循环的底层实现原理与调度算法
async/await协程的6种高级用法模式
异步HTTP请求的性能优化技巧（速度提升15倍+）
常见异步陷阱的26种解决方案

# 异步HTTP请求示例  
import aiohttp  
import asyncio  

async def fetch_async(urls):  
    async with aiohttp.ClientSession() as session:  
        tasks = [session.get(url) for url in urls]  
        responses = await asyncio.gather(*tasks)  
        return [await r.text() for r in responses]  

# 10个URL仅需2秒（所有请求并行）

性能对比：

指标	同步	异步
10请求耗时	20秒	2秒
CPU利用率	15%	85%
内存占用	低	中等

️ 第二章：事件循环——异步引擎的核心

2.1 事件循环架构解析

graph TD A[事件循环启动] –> B[任务队列] B –> C{队列空?} C –>|是| D[等待IO事件] C –>|否| E[取第一个任务] E –> F[执行到await] F –> G[挂起任务,注册回调] G –> B D –> H[IO就绪] H –> I[触发回调] I –> B

2.2 自定义事件循环实战

import uvloop  
import asyncio  

# 配置uvloop（比默认循环快30%）  
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())  

# 获取当前循环  
loop = asyncio.get_event_loop()  

# 手动调度协程  
async def task(name):  
    print(f"{name} start")  
    await asyncio.sleep(1)  
    print(f"{name} end")  

coro1 = task("A")  
coro2 = task("B")  

loop.run_until_complete(asyncio.gather(coro1, coro2))

🧵 第三章：协程与任务——异步的基本单元

3.1 协程的四种创建方式

# 方式1：async def  
async def simple_coro():  
    await asyncio.sleep(1)  

# 方式2：@asyncio.coroutine装饰器（旧式）  
@asyncio.coroutine  
def legacy_coro():  
    yield from asyncio.sleep(1)  

# 方式3：生成器协程  
def gen_coro():  
    yield from asyncio.sleep(1)  

# 方式4：async with上下文  
async with aiohttp.ClientSession() as session:  
    await session.get(url)

3.2 任务的高级控制

async def worker(q: asyncio.Queue):  
    while True:  
        item = await q.get()  
        try:  
            # 处理任务...  
        finally:  
            q.task_done()  

async def main():  
    q = asyncio.Queue(maxsize=100)  
    # 创建worker池  
    tasks = [asyncio.create_task(worker(q)) for _ in range(5)]  
    # 添加任务  
    for i in range(1000):  
        await q.put(i)  
    # 等待队列清空  
    await q.join()  
    # 取消worker  
    for t in tasks:  
        t.cancel()  
    await asyncio.gather(*tasks, return_exceptions=True)

第四章：异步HTTP请求实战

4.1 高性能爬虫设计

from bs4 import BeautifulSoup  
import aiohttp  

async def scrape_page(url):  
    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(10)) as session:  
        async with session.get(url) as response:  
            html = await response.text()  
            soup = BeautifulSoup(html, 'lxml')  
            # 解析逻辑...  
            return data  

async def scrape_all(urls):  
    sem = asyncio.Semaphore(20)  # 限制并发数  
    async def limited_scrape(url):  
        async with sem:  
            return await scrape_page(url)  
    return await asyncio.gather(*[limited_scrape(url) for url in urls])

4.2 与FastAPI的异步集成

from fastapi import FastAPI  
import httpx  

app = FastAPI()  

@app.get("/proxy")  
async def proxy_request(url: str):  
    async with httpx.AsyncClient() as client:  
        resp = await client.get(url)  
        return resp.json()

第五章：常见错误与调试

5.1 协程未执行（未await）

async def test():  
    print("Running")  

# 错误：直接调用协程  
test()  # 输出：RuntimeWarning: coroutine 'test' was never awaited  

# 正确执行方式：  
asyncio.run(test())

5.2 事件循环策略冲突

错误：RuntimeError: Event loop is closed  
解决方案：  
1. 确保使用async/await管理资源生命周期  
2. 避免在协程外创建ClientSession  
3. 显式关闭循环：  
   loop = asyncio.get_event_loop()  
   try:  
       loop.run_until_complete(main())  
   finally:  
       loop.close()

第六章：课后实战与测验

6.1 优化同步数据库访问

# 原同步代码（PostgreSQL）  
def query_users():  
    with psycopg2.connect(DSN) as conn:  
        cursor = conn.cursor()  
        cursor.execute("SELECT * FROM users")  
        return cursor.fetchall()  

# 任务：改写为异步版本（使用asyncpg）  
# 要求：  
# 1. 支持连接池  
# 2. 实现分页查询  
# 3. 处理查询超时

6.2 设计异步限流器

# 实现一个滑动窗口限流器  
class RateLimiter:  
    def __init__(self, rate=10, period=1):  
        self.rate = rate  
        self.period = period  

    async def __aenter__(self):  
        # 实现限流逻辑...  

    async def __aexit__(self, *args):  
        pass  

# 使用示例  
async def limited_api_call():  
    async with RateLimiter(100, 60):  # 每分钟最多100次  
        return await call_external_api()