7 个 Python 库让你的网络编程脱胎换骨
python约 7 分钟请求一多，最先崩的往往不是业务代码，是你自己写的那层“凑合能跑”的网络逻辑。
我见过不少 Python 项目，接口一开始就两个，requests.get() 一把梭，跑得还挺欢。结果后面加上重试、代理、超时、并发、WebSocket、抓包排查、异步回调，代码立马开始发散。最典型的味道就是：一个脚本里全是 try/except，超时和连接失败混着判，日志也只打一句 request failed，真到线上抖起来，根本不知道是 DNS 慢了，连接池满了，还是对端半天不回包。
Python 做网络编程，不是不能糊。就是糊久了，迟早得还。
下面这 7 个库，我不是按“功能分类”那种讲法来排的，没那个必要。我是按平时真会碰到的场景来放。哪个地方会让你少踩坑，哪个地方能让代码看起来像真干过活，不像 Demo，我就写哪个。
先说最常见的，httpx。
很多人还停在 requests，倒也不是不行，但你只要一碰异步接口批量调用，就知道它开始别扭了。httpx 这地方顺手得多，关键是同步异步接口风格接近，迁移的时候脑子不用拧两套。
比如批量探活内部服务，我一般会把超时拆开，不会只写一个总超时。连接超时和读取超时压根不是一回事。
import asyncio
import httpxasyncdefprobe(url: str):
timeout = httpx.Timeout(connect=1.0, read=2.0, write=2.0, pool=1.0)
asyncwith httpx.AsyncClient(timeout=timeout) as client:
try:
r = await client.get(url)
return url, r.status_code, r.text[:60]
except httpx.ConnectTimeout:
return url, "connect_timeout", ""
except httpx.ReadTimeout:
return url, "read_timeout", ""
except Exception as e:
return url, type(e).__name__, ""
asyncdefmain():
urls = [
"https://api.service-a.local/health",
"https://api.service-b.local/health",
]
result = await asyncio.gather(*(probe(u) for u in urls))
for item in result:
print(item)
asyncio.run(main())
这种代码拿去查线上问题是够用的。尤其你后面要补代理、证书校验、连接池限制，httpx 比较顺。
第二个我会放 aiohttp。
这玩意很多人只知道“异步 HTTP 客户端”，其实我更常拿它做轻量网关、回调接收器、小型 mock 服务。你排查第三方回调乱七八糟的时候，临时起个服务接住原始请求，比盯着文档猜强多了。
from aiohttp import webasyncdefwebhook(request: web.Request):
body = await request.text()
print("headers=", dict(request.headers))
print("body=", body)
return web.json_response({"ok": True})
app = web.Application()
app.router.add_post("/notify/payment", webhook)
web.run_app(app, host="0.0.0.0", port=9002)
这种代码不花哨，但真能救命。尤其对方说“我们已经推了，你们没收到”，你把这个服务挂上去，Nginx access log 再一对，问题基本就缩到很小了。
第三个，websockets。
只要你碰过实时推送、行情订阅、聊天室、设备状态上报，就知道 WebSocket 这东西表面简单，实际上最容易写成一坨。心跳没做好，断线重连没做好，现场就会出现一堆僵尸连接。
我一般至少会把重连和心跳写进去，不然线上断一次你就等着人工重启。
import asyncio
import json
import websocketsasyncdefconsume():
whileTrue:
try:
asyncwith websockets.connect("ws://127.0.0.1:8765/ws", ping_interval=20) as ws:
await ws.send(json.dumps({"type": "subscribe", "topic": "order_status"}))
asyncfor msg in ws:
print("recv:", msg)
except Exception as e:
print("ws reconnect:", type(e).__name__)
await asyncio.sleep(3)
asyncio.run(consume())
这里我第一眼最不信的，永远是“服务端不会断”。别信。网络只要稍微抖一下，你代码里那些理所当然的长连接，马上开始装死。
第四个，tenacity。
这个库不大，但很实用。很多人手写重试，越写越像事故现场：for i in range(3)，里面随便 sleep(1)，把所有异常都重试一遍。结果本来参数错了也重试，认证失败也重试，越重越炸。
tenacity 的好处是，哪些异常该重试，等多久，最多几次，都能写得清楚。
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import httpx@retry(
stop=stop_after_attempt(4),
wait=wait_exponential(multiplier=0.5, min=1, max=5),
retry=retry_if_exception_type((httpx.ConnectError, httpx.ReadTimeout)),
)
deffetch_user_profile(uid: int):
r = httpx.get(f"https://api.example.com/user/{uid}", timeout=2.0)
r.raise_for_status()
return r.json()
这种地方别偷懒。重试不是“失败了再来一次”这么简单，它本质上是在放大请求。你调用的是数据库网关、支付接口、库存服务，重试策略写糙了，就是你自己补刀。
第五个，scapy。
这个库一出来，味道就从“写业务”变成“查问题”了。我平时不建议一上来就抓包，但有些问题不看包，你猜一天也没用。比如 TCP 三次握手有没有完成，DNS 请求到底发没发出去，某些奇怪的 RST 是谁打的。
拿它做个简单探测就很直接。
from scapy.all import IP, TCP, sr1defcheck_port(host: str, port: int):
pkt = IP(dst=host) / TCP(dport=port, flags="S")
resp = sr1(pkt, timeout=1, verbose=0)
if resp isNone:
return"no_response"
if resp.haslayer(TCP):
flags = resp.getlayer(TCP).flags
if flags == 0x12:
return"open"
if flags == 0x14:
return"closed"
return"unknown"
print(check_port("10.10.20.8", 443))
这类代码别拿去当端口扫描器炫技，真正常用场景反而很朴素：服务说自己开着，你机器就是连不上，那就先看包，不要先怀疑人生。
第六个，pyzmq。
有些内部服务通信，HTTP 真不一定顺手。尤其是进程间快速分发消息、轻量任务投递、状态广播这些场景，pyzmq 比你自己拿 socket 硬写舒服多了。代码短，而且模型很清楚。
# sender.py
import zmq
import jsonctx = zmq.Context()
sock = ctx.socket(zmq.PUSH)
sock.bind("tcp://127.0.0.1:5557")
for i in range(5):
sock.send_string(json.dumps({"job_id": i, "type": "refresh_cache"}))
# worker.py
import zmq
import jsonctx = zmq.Context()
sock = ctx.socket(zmq.PULL)
sock.connect("tcp://127.0.0.1:5557")
whileTrue:
msg = json.loads(sock.recv_string())
print("handle", msg["job_id"], msg["type"])
这种东西特别适合你不想把一堆轻任务都塞进 Redis 或 MQ 的时候。不是说它能替代正式消息系统，但小范围用，挺利索。
第七个，asyncio 本尊。
很多人把它当标准库，不当“库”看，但网络编程里它就是骨架。前面那些异步 HTTP、WebSocket，本质上都绕不开它。问题也往往出在这：任务取消没处理、超时没包住、并发上限没控住，代码一多就开始隐性泄漏。
我一般会先把并发闸门加上，不然批量请求很容易把自己打死。
import asyncio
import httpxsem = asyncio.Semaphore(20)
asyncdeffetch(url: str):
asyncwith sem:
asyncwith httpx.AsyncClient() as client:
try:
r = await client.get(url, timeout=2.5)
return url, r.status_code
except Exception as e:
return url, type(e).__name__
asyncdefbatch_run():
urls = [f"https://api.example.com/item/{i}"for i in range(100)]
result = await asyncio.gather(*(fetch(u) for u in urls))
print(result[:10])
asyncio.run(batch_run())
这个 Semaphore 看着不起眼，线上特别有用。很多“对方服务怎么突然限流了”的锅，往回翻，都是自己一口气冲太猛。
这 7 个库放一起看，其实就两类活。
一类是把请求发得更稳，像 httpx、aiohttp、websockets、tenacity、asyncio。另一类是问题来了别瞎猜，像 scapy、pyzmq 这种，一个偏网络现场，一个偏通信模型。
我现在看一个 Python 网络项目，第一眼不是看它功能多不多，而是看这几件事：超时拆没拆，重试是不是乱打，并发有没有阀，长连接断了会不会自己活过来，日志里到底能不能看出是连接失败还是读超时。很多代码表面跑得挺顺，一上量就露馅，原因都不高级，基本就是这些细节没站住。
网络编程这活，花样很多，真值钱的反而不是“知道多少协议名词”，而是你出了问题以后，能不能尽快把范围缩小。库只是工具，顺手的工具会让排查少走很多弯路。这个差别，写两天 Demo 看不出来，线上抖两次，立刻就明白了。