python

Python 网络编程：socket、HTTP 客户端与服务端开发完全指南

Python 网络编程从底层到应用层：socket TCP/UDP 编程、HTTP 客户端（requests/httpx/urllib）、简易 Web 服务器、WebSocket 实时通信。包含完整可运行代码示例和常见问题排查。

spcent 7 分钟阅读 3181 字

网络编程是连接程序与世界的桥梁。从底层的 socket 到高层的 HTTP 请求，本文带你掌握 Python 中网络通信的每个层次。

目录

  1. 网络编程层次概览
  2. Socket 基础：TCP 客户端/服务端
  3. UDP 编程：无连接通信
  4. HTTP 客户端：requests vs urllib
  5. 异步 HTTP：aiohttp
  6. 用 Python 写 Web 服务器
  7. WebSocket 实时通信
  8. 网络调试技巧

1. 网络编程层次概览

应用层：HTTP、FTP、DNS
  │
传输层：TCP（可靠连接）、UDP（无连接）
  │
网络层：IP（寻址与路由）
  │
链路层：以太网、WiFi

Python 可在每个层次编程：

层次Python 工具难度
应用层（HTTP）requests, aiohttp, urllib⭐⭐
传输层socket⭐⭐⭐
底层scapy（数据包操作）⭐⭐⭐⭐

TCP 核心原理与连接建立

要写出高效可靠的网络程序，必须理解 TCP 的工作机制。TCP 是面向连接的可靠传输协议，其核心特性包括三次握手建立连接、四次挥手断开连接、滑动窗口流量控制和拥塞控制算法。

三次握手与四次挥手

客户端                    服务端
  │    SYN seq=x        │
  │ ─────────────────▶  │
  │                     │
  │  SYN seq=y, ACK=x+1 │
  │ ◀─────────────────  │
  │                     │
  │    ACK=y+1          │
  │ ─────────────────▶  │
  │    【连接建立】      │

Python 的 socket 模块在 connect() 和 accept() 时自动完成三次握手，开发者无需手动发送 SYN 包。但理解这个过程对于排查连接问题至关重要：如果服务端 SYN backlog 队列满，新的连接请求会被静默丢弃，客户端表现为 connect timeout。

滑动窗口与拥塞控制

TCP 通过滑动窗口机制实现流量控制（接收方告知发送方自己的缓冲能力）和拥塞控制（发送方根据网络拥塞程度动态调整发送速率）。现代 Linux 内核默认使用 CUBIC 拥塞控制算法，在高带宽延迟积网络（如跨国链路）中，BBR 算法表现更优：

# 查看当前拥塞控制算法
cat /sys/class/net/eth0/queues/rx-0/rps_cpus
sysctl net.ipv4.tcp_congestion_control

# 切换到 BBR（需要内核 4.9+）
sudo sysctl -w net.ipv4.tcp_congestion_control=bbr

调优建议：对于高并发服务端，net.ipv4.tcp_tw_reuse = 1 可复用 TIME_WAIT 状态的连接，但 tcp_tw_recycle 在 NAT 环境下会导致问题（已被内核 4.12+ 移除），应避免使用。

2. Socket 基础：TCP 客户端/服务端

2.1 TCP 服务端

import socket
import threading

def handle_client(conn, addr):
    """处理客户端连接"""
    print(f"客户端 {addr} 已连接")
    with conn:
        while True:
            data = conn.recv(1024)
            if not data:
                break
            print(f"收到: {data.decode()}")
            conn.sendall(b"Echo: " + data)
    print(f"客户端 {addr} 已断开")

def start_server(host="0.0.0.0", port=8888):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        s.listen(5)
        print(f"服务器启动于 {host}:{port}")
        
        while True:
            conn, addr = s.accept()
            thread = threading.Thread(target=handle_client, args=(conn, addr))
            thread.start()

if __name__ == "__main__":
    start_server()

2.2 TCP 客户端

import socket

def send_message(message, host="127.0.0.1", port=8888):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect((host, port))
        s.sendall(message.encode())
        response = s.recv(1024)
        print(f"响应: {response.decode()}")

send_message("Hello, Server!")

2.3 使用 selectors 实现高效 I/O（单线程多连接）

import selectors
import socket

sel = selectors.DefaultSelector()

def accept(sock):
    conn, addr = sock.accept()
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)

def read(conn):
    data = conn.recv(1024)
    if data:
        conn.send(data)
    else:
        sel.unregister(conn)
        conn.close()

server = socket.socket()
server.bind(("0.0.0.0", 8888))
server.listen()
server.setblocking(False)
sel.register(server, selectors.EVENT_READ, accept)

while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj)

3. UDP 编程：无连接通信

import socket

# UDP 服务端
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
    s.bind(("0.0.0.0", 9999))
    print("UDP 服务器启动")
    while True:
        data, addr = s.recvfrom(1024)
        print(f"来自 {addr}: {data.decode()}")
        s.sendto(b"Received: " + data, addr)

# UDP 客户端
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
    s.sendto(b"Hello UDP", ("127.0.0.1", 9999))
    data, _ = s.recvfrom(1024)
    print(data.decode())

4. HTTP 客户端：requests vs urllib

4.1 requests（最常用）

import requests

# GET
resp = requests.get("https://api.github.com/users/python")
print(resp.status_code)
print(resp.json()["public_repos"])

# 带参数
resp = requests.get("https://httpbin.org/get", params={"key": "value"})

# POST JSON
resp = requests.post(
    "https://httpbin.org/post",
    json={"name": "Alice", "age": 25},
    headers={"Authorization": "Bearer token123"},
)

# 下载文件
resp = requests.get("https://example.com/large.zip", stream=True)
with open("large.zip", "wb") as f:
    for chunk in resp.iter_content(chunk_size=8192):
        f.write(chunk)

# Session（保持 cookies/连接池）
with requests.Session() as s:
    s.headers.update({"User-Agent": "MyApp/1.0"})
    s.get("https://example.com/login")

4.2 urllib（标准库，无需安装）

import urllib.request
import urllib.parse

# GET
with urllib.request.urlopen("https://httpbin.org/get") as resp:
    print(resp.read().decode())

# POST
data = urllib.parse.urlencode({"key": "value"}).encode()
req = urllib.request.Request("https://httpbin.org/post", data=data, method="POST")
with urllib.request.urlopen(req) as resp:
    print(resp.read().decode())

4.3 httpx（现代化 HTTP 客户端）

# pip install httpx
import httpx

# 同步 API（兼容 requests）
resp = httpx.get("https://api.example.com/items")

# 异步 API（原生支持 async/await）
async with httpx.AsyncClient() as client:
    resp = await client.get("https://api.example.com/items")
    print(resp.json())

# HTTP/2 支持
client = httpx.Client(http2=True)

5. 异步 HTTP：aiohttp

# pip install aiohttp
import aiohttp
import asyncio

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

async def fetch_one(session, url):
    async with session.get(url) as resp:
        return await resp.text()

urls = ["https://httpbin.org/get"] * 5
results = asyncio.run(fetch_all(urls))
print(f"获取 {len(results)} 个页面")

6. 用 Python 写 Web 服务器

6.1 最简 HTTP 服务器

# 一行命令启动文件服务器
# python -m http.server 8000

# 程序化使用
from http.server import HTTPServer, SimpleHTTPRequestHandler

server = HTTPServer(("0.0.0.0", 8000), SimpleHTTPRequestHandler)
print("Serving on port 8000")
server.serve_forever()

6.2 自定义请求处理

from http.server import HTTPServer, BaseHTTPRequestHandler
import json

class APIHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/api/users":
            data = [{"id": 1, "name": "Alice"}]
            self._send_json(data)
        else:
            self.send_error(404)
    
    def do_POST(self):
        content_length = int(self.headers['Content-Length'])
        body = self.rfile.read(content_length)
        data = json.loads(body)
        
        self.send_response(201)
        self._send_json({"created": True, "data": data})
    
    def _send_json(self, data):
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(data).encode())

server = HTTPServer(("0.0.0.0", 8000), APIHandler)
server.serve_forever()

Python 网络编程性能优化

网络服务的性能瓶颈往往不在计算而在 I/O。理解不同 I/O 模型的适用场景，是写出高性能网络程序的关键。

I/O 模型Python 实现适用场景并发能力
阻塞 I/Othreading简单应用，低并发~1000 线程
非阻塞 I/Oselect/poll/epoll高并发连接，无计算万级连接
多路复用selectors现代标准，跨平台万级连接
异步 I/Oasyncio + aiohttp现代推荐方案十万级连接

线程池 vs 异步 I/O：线程池的硬伤是 GIL（全局解释器锁），纯 Python 代码无法真正并行；但 threading 处理 I/O 密集型任务时（如网络请求），GIL 会在 I/O 操作时释放，此时线程池仍然有效。对于 CPU 密集型 + 网络 I/O 混合场景，asyncio 配合线程池执行器是最佳组合：

import asyncio
from concurrent.futures import ThreadPoolExecutor

# CPU 密集型计算 + 网络 I/O
async def mixed_work():
    loop = asyncio.get_running_loop()
    
    # 网络 I/O：asyncio 原生
    http_task = asyncio.create_task(fetch_api())
    
    # CPU 密集型：放到线程池避免阻塞事件循环
    with ThreadPoolExecutor(max_workers=4) as pool:
        cpu_result = await loop.run_in_executor(
            pool,
            heavy_computation,
            data
        )
    
    api_result = await http_task
    return process(cpu_result, api_result)

Zero-Copy 与内存优化

对于文件传输等高吞吐量场景，Python 3.3+ 提供了 socket.sendfile() 实现零拷贝传输——数据直接从内核文件缓存发送到网卡，不经过用户空间：

import socket

def send_file_zero_copy(sock: socket.socket, file_path: str):
    """零拷贝文件传输，CPU 占用极低"""
    with open(file_path, 'rb') as f:
        # sendfile 直接将文件内容从内核推送到 socket
        # offset=None 表示从当前位置开始
        sock.sendfile(f)

适用条件：sendfile() 要求传输的是磁盘文件而非内存对象；对于加密传输（如 TLS），零拷贝不可用，因为数据需要先解密/加密才能发送。

7. WebSocket 实时通信

# pip install websockets
import asyncio
import websockets

async def handler(websocket, path):
    async for message in websocket:
        print(f"收到: {message}")
        await websocket.send(f"Echo: {message}")

async def main():
    async with websockets.serve(handler, "localhost", 8765):
        print("WebSocket server on ws://localhost:8765")
        await asyncio.Future()   # 永远运行

asyncio.run(main())

# 客户端
async def ws_client():
    async with websockets.connect("ws://localhost:8765") as ws:
        await ws.send("Hello WebSocket!")
        response = await ws.recv()
        print(response)

asyncio.run(ws_client())

网络协议安全：TLS/SSL 加密通信

生产环境的网络通信必须使用 TLS 加密，防止中间人攻击和数据窃听。Python 3.7+ 的 ssl 模块已经内置了对 OpenSSL 的完整封装。

TLS 服务端配置

import ssl
import socket

# 创建 TLS 上下文
context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
# 加载服务器证书和私钥
context.load_cert_chain(certfile='server.crt', keyfile='server.key')
# 强制要求客户端证书（双向 TLS）
# context.verify_mode = ssl.CERT_REQUIRED
# context.load_verify_locations('ca.crt')

# 包装普通 socket 为 TLS socket
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
    sock.bind(('0.0.0.0', 8443))
    sock.listen(5)
    conn, addr = sock.accept()
    with context.wrap_socket(conn, server_side=True) as tls_conn:
        data = tls_conn.recv(1024)
        tls_conn.sendall(b"Encrypted response")

TLS 安全配置检查清单

检查项推荐配置原因
协议版本TLS 1.2+TLS 1.0/1.1 已被主流浏览器弃用
加密套件ECDHE+AESGCM前向保密 + AEAD 认证加密
证书有效期≤ 397 天浏览器强制要求
HSTSmax-age=31536000防止 SSL stripping 攻击
OCSP Stapling启用减少首次握手延迟
# 快速检查服务端 TLS 配置
python -c "import ssl; print(ssl.OPENSSL_VERSION)"
# OpenSSL 3.0+ 支持 QUIC/TLS 1.3

最佳实践：使用 Let’s Encrypt 自动续期证书（certbot），配合 acme.sh 在 Python 应用中自动热加载新证书，无需重启服务。

生产环境部署与监控

将 Python 网络服务部署到生产环境需要考虑多个维度：进程管理、负载均衡、日志收集、健康检查。

多进程架构（绕过 GIL）

import os
import socket
from multiprocessing import Pool

def worker(sock):
    """每个进程独立监听套接字（SO_REUSEPORT 实现内核级负载均衡）"""
    while True:
        conn, addr = sock.accept()
        handle_client(conn, addr)

def run_multi_process(host='0.0.0.0', port=8888, workers=4):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # Linux 3.9+ 支持 SO_REUSEPORT：内核将新连接均匀分配给多个进程
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
    sock.bind((host, port))
    sock.listen(128)
    
    with Pool(workers) as pool:
        # 所有 worker 共享同一个监听 socket
        pool.map(worker, [sock] * workers)

if __name__ == '__main__':
    run_multi_process()

生产部署检查清单

  1. 进程管理：使用 systemd 或 supervisord，配置自动重启、日志轮转
  2. 反向代理：Nginx 前置处理静态文件、TLS 终止、限流、压缩
  3. 监控指标：Prometheus 采集连接数、请求延迟、错误率；Grafana 可视化
  4. 日志规范：结构化 JSON 日志，包含 trace_id 用于链路追踪
# Nginx 反向代理 WebSocket + HTTP
upstream backend {
    least_conn;  # 最少连接负载均衡
    server 127.0.0.1:8001;
    server 127.0.0.1:8002;
}

server {
    listen 443 ssl http2;
    ssl_certificate /etc/ssl/certs/server.crt;
    ssl_certificate_key /etc/ssl/private/server.key;
    
    location / {
        proxy_pass http://backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
    
    location /ws {
        proxy_pass http://backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_read_timeout 86400;
    }
}

8. 网络调试技巧

# 1. 查看 DNS 解析
import socket
print(socket.gethostbyname("example.com"))
print(socket.getaddrinfo("example.com", 80))

# 2. 获取本机 IP
hostname = socket.gethostname()
print(socket.gethostbyname(hostname))

# 3. 端口是否开放
sock = socket.socket()
result = sock.connect_ex(("localhost", 5432))
print("开放" if result == 0 else "关闭")

延伸阅读

网络编程的核心不在于记住 API，而在于理解 TCP 连接的三次握手、HTTP 的请求/响应模型、以及 I/O 类型（阻塞/非阻塞/异步）的差异。掌握这些，任何网络库都能快速上手。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

  1. 《Python高级编程》目录
  2. 《Python高级编程》11.3 PEP 流程与版本迁移策略
  3. 《Python高级编程》11.2 嵌入式与自由线程运行时