Python 网络编程与异步实战

当你的 Python 脚本需要同时请求十几个 API、维持几百个 WebSocket 连接时，同步阻塞模式会让性能直接崩塌。aiohttp 和 websockets 这两个库，让你在同一事件循环里并发处理成百上千个网络请求。本文从 aiohttp 客户端并发请求讲起，到 WebSocket 服务端与客户端的双向通信，最后覆盖连接池、限流与错误重试的生产级实践。

一、为什么要用异步网络编程

传统的 requests 库是同步阻塞的：发出请求后，程序会停在那里等服务器返回响应，期间 CPU 空闲。如果你要请求 100 个接口，串行执行就是 100 倍的延迟。用多线程虽然能解决并发问题，但线程切换有开销，而且 Python 的 GIL 让 CPU 密集型多线程并不能真正并行。

异步网络编程的核心思路是：在等待网络响应的间隙，把控制权交还给事件循环，去处理其他连接。这样单线程就能处理成百上千个并发连接，特别适合 I/O 密集型的网络应用。

# 同步：10 个请求串行，总耗时 = 10 × 单次延迟
import requests

urls = [f"https://api.example.com/data/{i}" for i in range(10)]
for url in urls:
    resp = requests.get(url)
    print(resp.status_code)

# 异步：10 个请求并发，总耗时 ≈ 单次延迟
import aiohttp, asyncio

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.json()

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

asyncio.run(main())

二、aiohttp 客户端实战

下面是一个更贴近真实场景的例子：并发请求多个 API，带超时、错误重试和结果聚合。

import aiohttp, asyncio

async def fetch_with_retry(session, url, retries=3, timeout=10):
    for attempt in range(retries):
        try:
            async with session.get(url, timeout=timeout) as resp:
                if resp.status == 200:
                    return await resp.json()
                else:
                    print(f"[{url}] 状态码 {resp.status}，第 {attempt+1} 次重试")
        except Exception as e:
            print(f"[{url}] 异常 {e}，第 {attempt+1} 次重试")
        await asyncio.sleep(1)
    return {"error": "failed", "url": url}

async def batch_fetch(urls):
    connector = aiohttp.TCPConnector(limit=20)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [fetch_with_retry(session, url) for url in urls]
        return await asyncio.gather(*tasks)

urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 21)]
results = asyncio.run(batch_fetch(urls))
print(f"成功获取 {len([r for r in results if 'error' not in r])} / {len(urls)} 条")

TCPConnector(limit=20) 控制同时打开的最大连接数，防止瞬间发起几百个连接把目标服务器打挂。这是生产环境必须设置的限流参数。

三、aiohttp Web 服务端

aiohttp 不仅是客户端，也能用来写异步 Web 服务。它的路由和视图风格类似 Flask，但底层是完全异步的。

from aiohttp import web
import asyncio

async def hello(request):
    name = request.match_info.get("name", "世界")
    await asyncio.sleep(0.01)  # 模拟异步处理
    return web.json_response({"message": f"你好，{name}"})

async def slow_handler(request):
    await asyncio.sleep(2)
    return web.json_response({"status": "done"})

app = web.Application()
app.add_routes([
    web.get("/", hello),
    web.get("/hello/{name}", hello),
    web.get("/slow", slow_handler),
])

if __name__ == "__main__":
    web.run_app(app, port=8080)

注意：/slow 接口需要 2 秒才能返回，但这期间服务器仍然能同时处理其他请求，因为 asyncio.sleep(2) 是异步等待，不会阻塞事件循环。这就是异步 Web 框架相比同步框架的核心优势。

四、WebSocket 双向通信

HTTP 是请求-响应模式，客户端必须先发起请求，服务器才能返回数据。但很多场景需要服务器主动推送消息：实时聊天、股票行情、在线协作文档。这时就需要 WebSocket。

用 websockets 库写一个最简单的 WebSocket 服务端：

import asyncio
import websockets

async def echo(websocket):
    async for message in websocket:
        print(f"收到：{message}")
        await websocket.send(f"服务器收到：{message}")

async def main():
    async with websockets.serve(echo, "localhost", 8765):
        await asyncio.Future()  # 永久运行

asyncio.run(main())

对应的客户端：

import asyncio
import websockets

async def client():
    async with websockets.connect("ws://localhost:8765") as ws:
        await ws.send("你好服务器")
        response = await ws.recv()
        print(f"服务器回复：{response}")

asyncio.run(client())

五、并发控制与信号量

当你有几千个 URL 需要抓取时，不能用 asyncio.gather 一次性全部发起——这会瞬间打开几千个连接。正确做法是用 asyncio.Semaphore 控制并发数。

import aiohttp, asyncio

sem = asyncio.Semaphore(10)  # 最多同时 10 个并发请求

async def fetch_limited(session, url):
    async with sem:  # 获取信号量
        try:
            async with session.get(url) as resp:
                return await resp.text()
        except Exception as e:
            return f"错误：{e}"

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_limited(session, url) for url in urls]
        return await asyncio.gather(*tasks)

urls = [f"https://example.com/{i}" for i in range(500)]
results = asyncio.run(main(urls))
print(f"完成 {len(results)} 个请求")

Semaphore(10) 保证同一时刻最多只有 10 个请求在飞行中，其余的在信号量前面排队等待。这是异步爬虫的标准写法。

六、常见坑与性能调优

坑一：在异步代码里调用同步阻塞函数。比如在 async def 里直接调用 requests.get() 或 time.sleep()，会直接阻塞整个事件循环。遇到必须用的同步库，用 asyncio.to_thread() 把它丢到线程池里执行。

import asyncio, requests

async def main():
    # 错误：阻塞事件循环
    # r = requests.get("https://example.com")

    # 正确：放到线程池
    r = await asyncio.to_thread(requests.get, "https://example.com")
    print(r.status_code)

坑二：忘记关闭 Session 或 WebSocket 连接。未关闭的连接会导致文件描述符泄漏，程序运行久了会报 Too many open files。用 async with 上下文管理器是最安全的方式。

坑三：事件循环策略不兼容 Windows。Windows 上默认的 Proactor 事件循环对某些 SSL 场景有 bug。Python 3.8+ 在 Windows 上会自动用 SelectorEventLoop 作为 Web 场景的默认值，一般不需要手动设置。

七、总结

aiohttp 和 websockets 是 Python 异步网络编程的两大主力库。前者覆盖了 HTTP 客户端与服务端，后者负责 WebSocket 双向通信。掌握三个核心模式：用 ClientSession 管理连接池、用 Semaphore 控制并发数、用 async with 保证资源正确释放，你就能写出既能扛住高并发又不会把自己玩崩的网络程序。