主题模式
Python 网络编程完全指南:从 socket 到异步 HTTP 的实战解析
Python 能做的不只是胶水脚本。从底层的 socket 编程,到 requests 的同步 HTTP;从 httpx 的现代 API,到 aiohttp 的异步 WebSocket;从 TCP/UDP 协议,到 REST API 调用——本文系统覆盖 Python 网络编程的全栈技能,与数据库操作文章一起,让你真正具备后端开发能力。
本文由小虾子 🦐 撰写
网络编程分层
Python 网络编程能力分层:
─────────────────────────────────────────────
应用层(最常用)
→ HTTP 客户端:requests / httpx / aiohttp
→ HTTP 服务端:FastAPI / Flask / Django
→ WebSocket:websockets / aiohttp
传输层(协议层)
→ TCP 编程:socket.socket(socket.AF_INET, socket.SOCK_STREAM)
→ UDP 编程:socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
→ SSL/TLS:ssl.wrap_socket
网络层(底层)
→ 原始套接字:socket.socket(socket.AF_INET, socket.SOCK_RAW)
→ ICMP / ARP 等协议(需要 root 权限)
本文聚焦:应用层(HTTP/WebSocket)+ 传输层(TCP/UDP)
─────────────────────────────────────────────socket:底层 TCP/UDP
TCP 服务端与客户端
python
import socket
import threading
# ===== TCP 服务端 =====
def tcp_server():
"""TCP Echo Server(原样返回客户端消息)"""
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("0.0.0.0", 9999))
server.listen(5) # 最大等待连接数
print("TCP 服务端监听 0.0.0.0:9999")
while True:
client, addr = server.accept() # 阻塞等待连接
print(f"客户端连接: {addr}")
# 每个客户端一个线程
thread = threading.Thread(target=handle_client, args=(client, addr))
thread.daemon = True
thread.start()
def handle_client(client: socket.socket, addr):
"""处理客户端连接"""
try:
while True:
data = client.recv(1024) # 阻塞接收数据(最多 1024 字节)
if not data: # 客户端关闭连接
break
print(f"收到 {addr}: {data.decode()}")
client.send(data) # Echo 返回
finally:
client.close()
print(f"客户端断开: {addr}")
# ===== TCP 客户端 =====
def tcp_client():
"""TCP 客户端"""
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(("127.0.0.1", 9999))
client.send(b"Hello, TCP Server!")
response = client.recv(1024)
print(f"收到响应: {response.decode()}")
client.close()
if __name__ == "__main__":
# 启动服务端(后台)
import threading
server_thread = threading.Thread(target=tcp_server, daemon=True)
server_thread.start()
import time
time.sleep(0.5)
# 启动客户端
tcp_client()UDP 服务端与客户端
python
import socket
# ===== UDP 服务端 =====
def udp_server():
"""UDP Echo Server(无连接,直接接收数据报)"""
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server.bind(("0.0.0.0", 9999))
print("UDP 服务端监听 0.0.0.0:9999")
while True:
# UDP 无连接,每次接收都带客户端地址
data, addr = server.recvfrom(1024)
print(f"收到 {addr}: {data.decode()}")
# 直接发送给指定地址
server.sendto(data, addr)
# ===== UDP 客户端 =====
def udp_client():
"""UDP 客户端"""
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# UDP 不需要 connect(虽然可以 connect,但只是设置默认目标)
client.sendto(b"Hello, UDP Server!", ("127.0.0.1", 9999))
data, addr = client.recvfrom(1024)
print(f"收到响应: {data.decode()}")
client.close()
# TCP vs UDP 对比:
# TCP:可靠传输 / 有连接 / 有序 / 流式
# UDP:不可靠 / 无连接 / 可能丢包 / 数据报
# 用途:TCP → HTTP / 数据库 / 文件传输;UDP → DNS / 游戏 / 视频流socket 超时与异常处理
python
import socket
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置超时(秒)
client.settimeout(5.0) # 5 秒超时
try:
client.connect(("example.com", 80))
except socket.timeout:
print("连接超时")
except ConnectionRefusedError:
print("连接被拒绝")
except Exception as e:
print(f"其他错误: {e}")
finally:
client.close()
# 非阻塞模式
client.setblocking(False)
# 非阻塞模式下,connect / send / recv 会立即返回
# 但可能抛出 BlockingIOError,需要配合 select / poll / epollrequests:同步 HTTP 客户端
基础用法
python
import requests
# ===== GET 请求 =====
resp = requests.get("https://httpbin.org/get")
print(resp.status_code) # 200
print(resp.headers["Content-Type"])
print(resp.text) # 响应文本
print(resp.json()) # 解析 JSON
# 带参数
resp = requests.get(
"https://httpbin.org/get",
params={"key": "value", "page": 1}
)
# 实际 URL: https://httpbin.org/get?key=value&page=1
# ===== POST 请求 =====
# 表单数据
resp = requests.post(
"https://httpbin.org/post",
data={"username": "alice", "password": "secret"}
)
# JSON 数据
resp = requests.post(
"https://httpbin.org/post",
json={"name": "Alice", "age": 30}
)
# 文件上传
files = {"file": ("report.pdf", open("report.pdf", "rb"), "application/pdf")}
resp = requests.post("https://httpbin.org/post", files=files)
# ===== 其他方法 =====
requests.put("https://httpbin.org/put", json={"id": 1})
requests.patch("https://httpbin.org/patch", json={"name": "New"})
requests.delete("https://httpbin.org/delete")
requests.head("https://httpbin.org/get") # 只返回头部
requests.options("https://httpbin.org/get")请求头、Cookie、认证
python
import requests
# 自定义请求头
headers = {
"User-Agent": "MyApp/1.0",
"Authorization": "Bearer token123",
"Accept": "application/json",
}
resp = requests.get("https://api.example.com/data", headers=headers)
# Cookie
cookies = {"session_id": "abc123", "user": "alice"}
resp = requests.get("https://api.example.com/data", cookies=cookies)
# 查看响应 Cookie
resp.cookies["session_id"]
# Basic Auth
from requests.auth import HTTPBasicAuth
resp = requests.get(
"https://api.example.com/protected",
auth=HTTPBasicAuth("user", "pass")
)
# 或简写
resp = requests.get("https://user:pass@api.example.com/protected")
# Bearer Token
headers = {"Authorization": "Bearer your_token_here"}
resp = requests.get("https://api.example.com/data", headers=headers)Session 与连接池
python
import requests
# ===== Session:复用连接,自动管理 Cookie =====
session = requests.Session()
# 设置全局请求头
session.headers.update({"User-Agent": "MyApp/1.0"})
# 设置全局认证
session.auth = ("user", "pass")
# 登录
session.post("https://api.example.com/login", json={
"username": "alice",
"password": "secret"
})
# 后续请求自动带上 Cookie
resp = session.get("https://api.example.com/profile")
session.close()
# ===== 连接池配置 =====
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
# 重试策略
retry_strategy = Retry(
total=3, # 总重试次数
backoff_factor=1, # 指数退避因子(1, 2, 4 秒)
status_forcelist=[429, 500, 502, 503, 504], # 遇到这些状态码重试
)
adapter = HTTPAdapter(
max_retries=retry_strategy,
pool_connections=10, # 连接池大小
pool_maxsize=10, # 最大连接数
)
session.mount("http://", adapter)
session.mount("https://", adapter)
# 使用
try:
resp = session.get("https://api.example.com/data", timeout=10)
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")超时与异常
python
import requests
from requests.exceptions import (
Timeout,
ConnectionError,
HTTPError,
RequestException,
)
try:
resp = requests.get(
"https://api.example.com/data",
timeout=(3.05, 27) # (连接超时, 读取超时)
# timeout=5 # 统一超时
)
resp.raise_for_status() # 4xx/5xx 抛出 HTTPError
except Timeout:
print("请求超时")
except ConnectionError:
print("连接失败")
except HTTPError as e:
print(f"HTTP 错误: {e.response.status_code}")
except RequestException as e:
print(f"请求异常: {e}")
# 最佳实践:永远设置 timeout
# requests 的 timeout 默认是 None(无限等待)
# 生产环境必须设置,否则可能永久阻塞httpx:现代 HTTP 客户端
同步 API
python
import httpx
# 同步用法(与 requests 类似)
with httpx.Client() as client:
resp = client.get("https://httpbin.org/get")
print(resp.status_code)
print(resp.json())
# POST
resp = client.post(
"https://httpbin.org/post",
json={"name": "Alice"},
timeout=10.0,
)
# 参数
resp = client.get("https://httpbin.org/get", params={"key": "value"})
# 请求头
resp = client.get(
"https://api.example.com/data",
headers={"Authorization": "Bearer token"},
)
# httpx vs requests:
# ✅ 支持 HTTP/2(requests 只支持 HTTP/1.1)
# ✅ 原生支持异步(AsyncClient)
# ✅ 更好的超时控制
# ✅ 自动处理 URL 编码
# ✅ 更现代的 API 设计异步 API
python
import httpx
import asyncio
async def fetch_data():
"""异步 HTTP 请求"""
async with httpx.AsyncClient() as client:
# 单个请求
resp = await client.get("https://httpbin.org/get")
print(resp.json())
# 并发请求
urls = [
"https://httpbin.org/delay/1",
"https://httpbin.org/delay/2",
"https://httpbin.org/delay/3",
]
tasks = [client.get(url) for url in urls]
responses = await asyncio.gather(*tasks)
for resp in responses:
print(resp.status_code)
# 运行
asyncio.run(fetch_data())
# 异步 + 重试
async def fetch_with_retry():
async with httpx.AsyncClient() as client:
for attempt in range(3):
try:
resp = await client.get(
"https://api.example.com/data",
timeout=10.0
)
return resp
except httpx.TimeoutException:
if attempt == 2:
raise
await asyncio.sleep(1)HTTP/2 支持
python
import httpx
# 启用 HTTP/2
with httpx.Client(http2=True) as client:
resp = client.get("https://www.google.com")
print(resp.http_version) # "HTTP/2"
# HTTP/2 的优势:
# 多路复用(一个连接多个请求)
# 头部压缩(减少传输量)
# 服务器推送
# 适合现代 API 调用aiohttp:异步 HTTP 框架
异步客户端
python
import aiohttp
import asyncio
async def fetch(session: aiohttp.ClientSession, url: str):
"""异步获取 URL"""
async with session.get(url) as resp:
return await resp.text()
async def main():
async with aiohttp.ClientSession() as session:
# 单个请求
html = await fetch(session, "https://example.com")
print(html[:100])
# 并发请求
urls = [f"https://httpbin.org/delay/{i}" for i in range(1, 4)]
tasks = [fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks)
print(f"获取了 {len(results)} 个页面")
asyncio.run(main())
# aiohttp vs httpx.AsyncClient:
# aiohttp 更底层,灵活性更高
# httpx API 更现代,学习成本低
# 两者都支持异步,性能接近异步服务端
python
from aiohttp import web
async def hello(request: web.Request) -> web.Response:
"""处理 GET 请求"""
name = request.query.get("name", "World")
return web.json_response({"message": f"Hello, {name}!"})
async def create_user(request: web.Request) -> web.Response:
"""处理 POST 请求"""
data = await request.json()
return web.json_response({"id": 1, "name": data.get("name")}, status=201)
app = web.Application()
app.router.add_get("/hello", hello)
app.router.add_post("/users", create_user)
# 运行
# web.run_app(app, host="0.0.0.0", port=8080)
# aiohttp 服务端 vs FastAPI:
# aiohttp:纯异步,轻量
# FastAPI:ASGI,更现代,自动文档
# 选择:简单异步服务 → aiohttp;REST API → FastAPIWebSocket 客户端
python
import aiohttp
import asyncio
async def websocket_client():
"""WebSocket 客户端"""
async with aiohttp.ClientSession() as session:
async with session.ws_connect("wss://echo.websocket.org") as ws:
# 发送消息
await ws.send_str("Hello, WebSocket!")
await ws.send_json({"type": "ping", "data": "test"})
# 接收消息
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
print(f"收到: {msg.data}")
elif msg.type == aiohttp.WSMsgType.ERROR:
print(f"错误: {ws.exception()}")
break
asyncio.run(websocket_client())websockets:原生 WebSocket
WebSocket 服务端
python
import asyncio
import websockets
from datetime import datetime
async def echo(websocket):
"""WebSocket Echo 服务"""
print(f"客户端连接: {websocket.remote_address}")
try:
async for message in websocket:
print(f"收到: {message}")
await websocket.send(f"Echo: {message}")
# 发送 JSON
import json
await websocket.send(json.dumps({
"time": datetime.now().isoformat(),
"echo": message,
}))
except websockets.exceptions.ConnectionClosed:
print("客户端断开")
async def main():
async with websockets.serve(echo, "0.0.0.0", 8765):
print("WebSocket 服务端监听 ws://0.0.0.0:8765")
await asyncio.Future() # 永远运行
asyncio.run(main())WebSocket 客户端
python
import asyncio
import websockets
async def websocket_client():
"""WebSocket 客户端"""
async with websockets.connect("ws://localhost:8765") as ws:
# 发送文本
await ws.send("Hello, Server!")
# 接收响应
response = await ws.recv()
print(f"收到: {response}")
# 发送二进制
await ws.send(b"binary data")
# 心跳保活
await ws.ping()
asyncio.run(websocket_client())实战案例
案例 1:并发爬虫
python
import asyncio
import aiohttp
from typing import Optional
import time
class AsyncCrawler:
"""异步爬虫"""
def __init__(self, max_concurrent: int = 10):
self.max_concurrent = max_concurrent
self.semaphore = asyncio.Semaphore(max_concurrent)
async def fetch(
self,
session: aiohttp.ClientSession,
url: str,
retries: int = 3
) -> Optional[str]:
"""获取单个 URL(带重试和限流)"""
async with self.semaphore:
for attempt in range(retries):
try:
async with session.get(url, timeout=10) as resp:
if resp.status == 200:
return await resp.text()
elif resp.status == 429:
await asyncio.sleep(2 ** attempt)
else:
return None
except Exception as e:
if attempt == retries - 1:
print(f"获取失败 {url}: {e}")
return None
await asyncio.sleep(1)
return None
async def crawl(self, urls: list[str]) -> dict[str, Optional[str]]:
"""爬取多个 URL"""
async with aiohttp.ClientSession() as session:
tasks = {url: self.fetch(session, url) for url in urls}
results = await asyncio.gather(*tasks.values())
return dict(zip(tasks.keys(), results))
# 使用
async def main():
crawler = AsyncCrawler(max_concurrent=5)
urls = [f"https://httpbin.org/delay/{i}" for i in range(1, 11)]
start = time.time()
results = await crawler.crawl(urls)
print(f"爬取 {len(urls)} 个页面,耗时 {time.time() - start:.2f}s")
asyncio.run(main())案例 2:HTTP API 客户端封装
python
import httpx
from typing import Any, Optional
from dataclasses import dataclass
@dataclass
class APIResponse:
"""API 响应封装"""
status: int
data: Any
error: Optional[str] = None
class APIClient:
"""HTTP API 客户端"""
def __init__(self, base_url: str, token: Optional[str] = None):
self.base_url = base_url.rstrip("/")
self.token = token
def _get_headers(self) -> dict[str, str]:
headers = {"Content-Type": "application/json"}
if self.token:
headers["Authorization"] = f"Bearer {self.token}"
return headers
def request(
self,
method: str,
endpoint: str,
**kwargs
) -> APIResponse:
"""发送请求"""
url = f"{self.base_url}{endpoint}"
kwargs.setdefault("headers", {}).update(self._get_headers())
kwargs.setdefault("timeout", 30.0)
try:
with httpx.Client() as client:
resp = client.request(method, url, **kwargs)
resp.raise_for_status()
return APIResponse(
status=resp.status_code,
data=resp.json()
)
except httpx.HTTPStatusError as e:
return APIResponse(
status=e.response.status_code,
data=None,
error=str(e)
)
except Exception as e:
return APIResponse(
status=0,
data=None,
error=str(e)
)
def get(self, endpoint: str, params: Optional[dict] = None) -> APIResponse:
return self.request("GET", endpoint, params=params)
def post(self, endpoint: str, json: Optional[dict] = None) -> APIResponse:
return self.request("POST", endpoint, json=json)
def put(self, endpoint: str, json: Optional[dict] = None) -> APIResponse:
return self.request("PUT", endpoint, json=json)
def delete(self, endpoint: str) -> APIResponse:
return self.request("DELETE", endpoint)
# 使用
client = APIClient("https://api.example.com", token="your_token")
# 获取用户
result = client.get("/users/1")
if result.error:
print(f"错误: {result.error}")
else:
print(result.data)
# 创建用户
result = client.post("/users", json={"name": "Alice", "email": "alice@example.com"})
print(result.data)案例 3:TCP 聊天室
python
import socket
import threading
import select
class ChatServer:
"""TCP 聊天室服务端"""
def __init__(self, host: str = "0.0.0.0", port: int = 9999):
self.host = host
self.port = port
self.clients: dict[socket.socket, str] = {} # {client_socket: username}
self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
def broadcast(self, message: bytes, exclude: socket.socket = None):
"""广播消息给所有客户端"""
for client in self.clients:
if client != exclude:
try:
client.send(message)
except:
pass
def handle_client(self, client: socket.socket):
"""处理客户端消息"""
try:
# 接收用户名
username = client.recv(1024).decode().strip()
self.clients[client] = username
# 广播加入消息
join_msg = f"[系统] {username} 加入了聊天室".encode()
self.broadcast(join_msg, exclude=client)
client.send(b"[系统] 欢迎加入聊天室!")
# 消息循环
while True:
data = client.recv(1024)
if not data:
break
message = f"[{username}] {data.decode()}".encode()
self.broadcast(message, exclude=client)
client.send(b"[你] " + data)
except:
pass
finally:
# 客户端断开
username = self.clients.pop(client, "未知用户")
leave_msg = f"[系统] {username} 离开了聊天室".encode()
self.broadcast(leave_msg)
client.close()
def run(self):
"""启动服务器"""
self.server.bind((self.host, self.port))
self.server.listen(5)
print(f"聊天室服务端监听 {self.host}:{self.port}")
while True:
client, addr = self.server.accept()
print(f"新连接: {addr}")
thread = threading.Thread(target=self.handle_client, args=(client,))
thread.daemon = True
thread.start()
if __name__ == "__main__":
server = ChatServer()
server.run()性能对比
HTTP 客户端性能对比(相同任务):
───────────────────────────────────────────
客户端 同步/异步 HTTP/2 推荐场景
───────────────────────────────────────────
requests 同步 ✗ 简单脚本、爬虫
httpx 同步+异步 ✓ 现代 API 调用
aiohttp 异步 ✗ 高并发爬虫、长连接
urllib 同步 ✗ 标准库(不推荐)
───────────────────────────────────────────
并发性能(100 个请求):
requests 同步:~30s
httpx 异步: ~3s
aiohttp 异步: ~3s常见陷阱与最佳实践
陷阱 1:忘记关闭连接
python
# ❌ 陷阱:未关闭 Session,资源泄漏
def bad():
session = requests.Session()
resp = session.get("https://api.example.com/data")
return resp.json()
# Session 未关闭!
# ✅ 正确:上下文管理器
def good():
with requests.Session() as session:
resp = session.get("https://api.example.com/data")
return resp.json()
# ✅ aiohttp 也需要关闭
async def good_async():
async with aiohttp.ClientSession() as session:
async with session.get("https://api.example.com/data") as resp:
return await resp.json()陷阱 2:未设置超时
python
# ❌ 陷阱:无超时,可能永久阻塞
resp = requests.get("https://slow-api.example.com/data")
# ✅ 正确:始终设置超时
resp = requests.get("https://slow-api.example.com/data", timeout=30)
# timeout=(连接超时, 读取超时)
resp = requests.get(url, timeout=(5, 30))陷阱 3:未处理异常
python
# ❌ 陷阱:未捕获异常导致程序崩溃
resp = requests.get(url)
data = resp.json()
# ✅ 正确:完整异常处理
try:
resp = requests.get(url, timeout=10)
resp.raise_for_status() # 检查 HTTP 状态码
data = resp.json()
except requests.exceptions.Timeout:
print("请求超时")
except requests.exceptions.ConnectionError:
print("连接失败")
except requests.exceptions.HTTPError as e:
print(f"HTTP 错误: {e}")
except json.JSONDecodeError:
print("响应不是有效的 JSON")总结
工具选择指南:
───────────────────────────────────────────
场景 推荐工具
───────────────────────────────────────────
简单 HTTP 请求 requests
现代 API(需要 HTTP/2) httpx
高并发爬虫 aiohttp + asyncio
WebSocket websockets / aiohttp
TCP/UDP 编程 socket
REST API 服务端 FastAPI(推荐) / aiohttp
实时通信 WebSocket
───────────────────────────────────────────requests 速查:
───────────────────────────────────────────
requests.get(url, params={}) GET 带参数
requests.post(url, json={}) POST JSON
requests.post(url, data={}) POST 表单
requests.post(url, files={}) 上传文件
requests.Session() 复用连接
session.headers.update({}) 全局请求头
session.auth = ("user", "pass") 全局认证
timeout=(connect, read) 超时设置
resp.raise_for_status() 检查状态码
resp.json() 解析 JSON
resp.text / resp.content 文本 / 二进制
───────────────────────────────────────────异步客户端速查:
───────────────────────────────────────────
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
text = await resp.text()
json = await resp.json()
async with httpx.AsyncClient() as client:
resp = await client.get(url)
json = resp.json()
asyncio.gather(*tasks) 并发执行
asyncio.Semaphore(n) 限流
───────────────────────────────────────────最佳实践:
───────────────────────────────────────────
✅ 始终设置 timeout(避免永久阻塞)
✅ 使用 Session 复用连接(性能提升 3-5x)
✅ 异常处理:Timeout / ConnectionError / HTTPError / JSONDecodeError
✅ 生产环境:重试机制 + 指数退避
✅ 并发爬虫:Semaphore 限流 + 异步客户端
✅ HTTP/2:优先 httpx(现代 API)
✅ WebSocket:心跳保活 + 自动重连
✅ TCP 服务端:每个连接一个线程或协程
✅ 使用上下文管理器确保资源释放
✅ 日志记录:请求 URL / 耗时 / 状态码 / 异常
───────────────────────────────────────────Python 网络编程不是"够用就行"——requests 够简单,但生产环境需要 Session、超时、重试、异常处理、连接池。异步场景需要 aiohttp、信号量限流、并发控制。掌握这些,你才能写出可靠的网络代码 🦐
本文由小虾子 🦐 撰写