Python异步编程：高性能网络应用实战

协程 (Coroutine)

• 轻量级线程，通过 async def 定义。
• 使用 await 暂停执行直到I/O操作完成，释放CPU资源。

async def fetch_data(url):
response = await aiohttp.get(url)
return response.text()

事件循环 (Event Loop)

• 调度协程的执行，监控I/O事件。
• 示例：

import asyncio

async def main():
task1 = asyncio.create_task(fetch_data("url1"))
task2 = asyncio.create_task(fetch_data("url2"))
await asyncio.gather(task1, task2)

asyncio.run(main())

HTTP服务：FastAPI + Uvicorn

• FastAPI支持异步端点：

from fastapi import FastAPI
app = FastAPI()

@app.get("/data")
async def get_data():
result = await fetch_data("https://api.example.com")
return {"data": result}

• 部署命令：uvicorn main:app –workers 4

TCP/WebSocket：aiohttp

• 异步HTTP客户端/服务器：

from aiohttp import web

async def websocket_handler(request):
ws = web.WebSocketResponse()
await ws.prepare(request)
async for msg in ws:
if msg.type == web.WSMsgType.TEXT:
await ws.send_str(f"Echo: {msg.data}")

连接池管理

• 复用数据库/HTTP连接：

async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
…

超时与重试

• 防止协程永久阻塞：

try:
await asyncio.wait_for(fetch_data(), timeout=5.0)
except asyncio.TimeoutError:
print("Request timed out")

使用uvloop替代默认事件循环

• 性能提升2-4倍：

import uvloop
uvloop.install()