7 个 Python 库让你的网络编程脱胎换骨

线上接口卡住了，日志里只剩一句：

ReadTimeout: HTTPSConnectionPool(host='api.xxx.com', port=443)

这种问题我一般不先怀疑业务代码。网络编程里最烦的不是“连不上”，而是“偶尔连不上、偶尔超时、偶尔粘包、偶尔一堆连接挂着不释放”。

Python 写网络程序，别只会 requests.get()。面试里问 7 个网络库，其实不是让你背名字，是看你知不知道不同问题该拿什么工具拆。

1、socket

socket 是底层货，绕不开。

只要你写 TCP、UDP，最后基本都绕到它身上。HTTP 再高级，底下也是 socket。

我面试时很喜欢问一个问题：TCP 一次 recv(1024) 能不能保证读到完整业务包？

不能。

TCP 是字节流，不管你上层业务边界。你发两次，对方可能一次收到；你发一次，对方也可能分几次收到。这个坑不踩一次，很难真记住。

import socket
import struct
defread_exact(conn, size):
    buf = bytearray()
while len(buf) < size:
        chunk = conn.recv(size - len(buf))
ifnot chunk:
raise ConnectionError("client closed before packet completed")
        buf.extend(chunk)
return bytes(buf)
defread_packet(conn):
    header = read_exact(conn, 4)
    body_len = struct.unpack("!I", header)[0]
if body_len > 2 * 1024 * 1024:
raise ValueError(f"packet too large: {body_len}")
return read_exact(conn, body_len)

这种代码看着土，但线上比花哨封装可靠。先读长度，再按长度读 body。别指望一次 recv 解决所有问题。

2、selectors

如果只会一个连接一个线程，连接数稍微上来，线程先把机器拖死。

selectors 是 Python 标准库里做 IO 多路复用的东西。Linux 下通常走 epoll，macOS 下走 kqueue，Windows 下也有对应实现。它适合写轻量 TCP 服务，尤其是你想理解事件驱动到底怎么回事的时候。

import selectors
import socket
sel = selectors.DefaultSelector()
defaccept(server):
    conn, addr = server.accept()
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read_client)
    print("new conn:", addr)
defread_client(conn):
    data = conn.recv(4096)
ifnot data:
        sel.unregister(conn)
        conn.close()
return
    conn.sendall(b"ok:" + data)
server = socket.socket()
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("0.0.0.0", 9001))
server.listen()
server.setblocking(False)
sel.register(server, selectors.EVENT_READ, accept)
whileTrue:
for key, _ in sel.select(timeout=1):
        key.data(key.fileobj)

这段代码不优雅，但它把网络编程最关键的东西摊开了：连接、事件、读写、关闭。

3、asyncio

asyncio 解决的是高并发 IO 等待问题。

很多人一看 async/await 就觉得高级，其实别神化。它不是让 CPU 跑得更快，而是让你在等待网络返回时，别傻站着。

适合什么？爬接口、批量调用内部服务、长连接网关、推送系统。

import asyncio
asyncdefcall_device(device_id):
    reader, writer = await asyncio.open_connection("127.0.0.1", 9001)
    payload = f"PING {device_id}".encode()
    writer.write(len(payload).to_bytes(4, "big") + payload)
await writer.drain()
    size = int.from_bytes(await reader.readexactly(4), "big")
    body = await reader.readexactly(size)
    writer.close()
await writer.wait_closed()
return device_id, body.decode()
asyncdefmain():
    tasks = [call_device(f"d{i}") for i in range(100)]
for item inawait asyncio.gather(*tasks, return_exceptions=True):
        print(item)
asyncio.run(main())

这里我会特别看 readexactly。能写出这个，比只会 await reader.read(1024) 靠谱一点。

4、aiohttp

asyncio 管事件循环，aiohttp 更贴近 HTTP 场景。

比如你要批量调用 500 个设备接口，用同步请求一个个打，超时的时候会很难看。aiohttp 可以配合连接池、超时、并发控制一起用。

import asyncio
import aiohttp
asyncdeffetch_status(session, device_id):
    url = f"http://ops-api.local/device/{device_id}/status"
try:
asyncwith session.get(url, timeout=3) as resp:
if resp.status != 200:
return device_id, "bad_status", resp.status
return device_id, "ok", await resp.json()
except asyncio.TimeoutError:
return device_id, "timeout", None
asyncdefscan_devices(ids):
    connector = aiohttp.TCPConnector(limit=80, limit_per_host=20)
asyncwith aiohttp.ClientSession(connector=connector) as session:
returnawait asyncio.gather(*(fetch_status session, i) for i in ids)

上面代码有个地方故意值得看：并发不是越大越好。limit 不设，打挂对方服务也不是没见过。

不过最后一行少了括号，应该这样写：

returnawait asyncio.gather(*(fetch_status(session, i) for i in ids))

这类小错，面试现场也挺常见。能自己发现，印象反而更好。

5、httpx

requests 好用，但它偏同步。httpx 的好处是同步、异步都支持，迁移成本低。

我一般在内部工具里用 httpx，尤其是要统一超时、重试、header 的地方。

import httpx
client = httpx.Client(
    timeout=httpx.Timeout(connect=2, read=5, write=3, pool=2),
    headers={"User-Agent": "ops-checker/1.0"}
)
defcheck_order(order_id):
    r = client.get(f"https://api.local/orders/{order_id}")
    r.raise_for_status()
return r.json()

这里别只写一个 timeout=5。连接超时、读取超时、连接池等待超时，最好分开。线上排查超时时，这几个不是一回事。

6、websockets

WebSocket 场景现在不少，监控大屏、设备实时状态、IM、日志推送都会用。

websockets 库比较直接，适合快速写一个长连接服务。

import asyncio
import websockets
clients = set()
asyncdefhandle(ws):
    clients.add(ws)
try:
asyncfor msg in ws:
if msg == "ping":
await ws.send("pong")
finally:
        clients.discard(ws)
asyncdefpush_alarm(text):
    dead = []
for ws in clients:
try:
await ws.send(text)
except Exception:
            dead.append(ws)
for ws in dead:
        clients.discard(ws)
asyncdefmain():
asyncwith websockets.serve(handle, "0.0.0.0", 9100):
await asyncio.Future()
asyncio.run(main())

长连接别忘了清理断开的客户端。不清理，时间一长，集合里全是尸体连接。

7、scapy

scapy 不太像业务库，更像排障工具。

抓包、构造包、看 TCP/UDP 细节，它很有用。一般业务开发用不到，但你真遇到“服务端说没收到，客户端说发了”的问题，抓包比吵架好使。

from scapy.all import sniff, TCP, IP
defshow(pkt):
if IP in pkt and TCP in pkt:
        print(
            pkt[IP].src,
"->",
            pkt[IP].dst,
"sport=", pkt[TCP].sport,
"dport=", pkt[TCP].dport,
"flags=", pkt[TCP].flags
        )
sniff(filter="tcp port 9001", prn=show, store=False)

这个库别乱用在生产机上长时间跑。抓包本身也吃资源，排查完就停。

这 7 个库放在一起看，顺序其实很清楚。

socket 让你知道底层怎么收发数据。selectors 让你知道多连接怎么扛。asyncio 让你知道 IO 等待怎么让出去。aiohttp 和 httpx 解决 HTTP 客户端/服务端里的实际问题。websockets 处理长连接。scapy 留给那些日志解释不了的现场。

面试时别只背库名。能讲清楚一次 recv 为什么读不完整、超时为什么要拆开配、长连接为什么要清理、批量请求为什么要限并发，这题基本就稳了。