Python 网络编程完全指南：Socket / HTTP / aiohttp / WebSocket 从入门到实战

Python 网络编程是构建网络应用、爬虫、API 客户端、实时通信系统的核心技能。从底层 Socket 到高层 HTTP 客户端库，再到异步编程和 WebSocket，掌握网络编程能极大拓展 Python 的应用范围。

本文全面讲解 Python 网络编程核心知识：

网络基础：OSI 模型与 TCP/IP 协议栈
Socket 编程：TCP/UDP 服务端与客户端实现
HTTP 客户端：urllib / requests / httpx 对比与实战
异步网络：aiohttp + asyncio 高并发请求
WebSocket 编程：实时双向通信实现
实战项目：构建高效网络请求工具

一、网络编程基础#

1.1 OSI 七层模型与 TCP/IP 协议栈#

1
OSI 七层模型            TCP/IP 四层模型        常见协议
2
─────────────────────────────────────────────────────────
3
7. 应用层              4. 应用层             HTTP, HTTPS, FTP, SMTP, DNS, WebSocket
4
6. 表示层              ↑                    SSL/TLS, JPEG, ASCII
5
5. 会话层              ↑                    RPC, NetBIOS
6
4. 传输层              3. 传输层             TCP, UDP, QUIC, SCTP
7
3. 网络层              2. 网络层             IP, ICMP, ARP
8
2. 数据链路层          1. 网络接口层         Ethernet, WiFi (802.11), PPP
9
1. 物理层              ↑                    光纤, 双绞线, 无线信号

1
# 网络编程中最常接触的是传输层和应用层
2
# Python socket 模块工作在传输层（TCP/UDP）
3
# requests/aiohttp 工作在应用层（HTTP/HTTPS）
4

5
import socket
6

7
# 查看本机协议支持
8
print(socket.has_ipv6)  # 是否支持 IPv6
9
print(socket.IPPROTO_TCP)  # TCP 协议编号
10
print(socket.IPPROTO_UDP)  # UDP 协议编号

1.2 TCP 三次握手与四次挥手#

1
# TCP 连接建立（三次握手）
2
"""
3
客户端 → SYN → 服务端
4
客户端 ← SYN+ACK ← 服务端
5
客户端 → ACK → 服务端
6
# 连接建立，可以开始传输数据
7

8
TCP 连接关闭（四次挥手）
9
客户端 → FIN → 服务端
10
客户端 ← ACK ← 服务端  （服务端进入半关闭状态）
11
客户端 ← FIN ← 服务端
12
客户端 → ACK → 服务端
13
# 连接完全关闭
14
"""
15

16
# 使用 socket 模块模拟 TCP 连接
17
import socket
18
import struct
19

20
# TCP 套接字创建
21
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
22
# AF_INET: IPv4, SOCK_STREAM: TCP
23
# AF_INET6: IPv6, SOCK_DGRAM: UDP

1.3 常用网络端口速查#

1
端口号    协议    用途
2
─────────────────────────────
3
20       FTP-DATA 文件传输数据
4
21       FTP     文件传输控制
5
22       SSH     安全远程登录
6
23       Telnet  远程登录
7
25       SMTP    邮件发送
8
53       DNS     域名解析
9
80       HTTP    超文本传输
10
110      POP3    邮件接收
11
143      IMAP    邮件访问
12
443      HTTPS   加密 HTTP
13
3000     常用开发端口
14
3306     MySQL   数据库
15
6379     Redis   缓存
16
8080     HTTP 备用端口
17
8443     HTTPS 备用端口

二、Socket 编程基础#

2.1 TCP 服务端实现#

1
import socket
2
import threading
3
import time
4

5
class TCPServer:
6
    def __init__(self, host='0.0.0.0', port=8888):
7
        self.host = host
8
        self.port = port
9
        self.server_socket = None
10
        self.clients = []
11

12
    def start(self):
13
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
14
        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
15
        self.server_socket.bind((self.host, self.port))
16
        self.server_socket.listen(5)
17
        print(f"[服务端] 监听 {self.host}:{self.port}")
18

19
        try:
20
            while True:
21
                client_socket, addr = self.server_socket.accept()
22
                self.clients.append(client_socket)
23
                print(f"[服务端] 新连接: {addr}")
24
                thread = threading.Thread(
25
                    target=self.handle_client,
26
                    args=(client_socket, addr),
27
                    daemon=True
28
                )
29
                thread.start()
30
        except KeyboardInterrupt:
31
            self.stop()
32

33
    def handle_client(self, client_socket, addr):
34
        try:
35
            while True:
36
                data = client_socket.recv(4096)
37
                if not data:
38
                    break
39
                message = data.decode('utf-8')
40
                print(f"[客户端 {addr}] {message}")
41
                response = f"服务器收到: {message}"
42
                client_socket.sendall(response.encode('utf-8'))
43
        except ConnectionResetError:
44
            print(f"[客户端 {addr}] 断开连接")
45
        finally:
46
            client_socket.close()
47
            self.clients.remove(client_socket)
48

49
    def stop(self):
50
        for client in self.clients:
51
            client.close()
52
        if self.server_socket:
53
            self.server_socket.close()
54
        print("[服务端] 已关闭")
55

56
if __name__ == '__main__':
57
    server = TCPServer()
58
    server.start()

2.2 TCP 客户端实现#

1
import socket
2
import sys
3

4
class TCPClient:
5
    def __init__(self, host='127.0.0.1', port=8888):
6
        self.host = host
7
        self.port = port
8
        self.client_socket = None
9

10
    def connect(self):
11
        self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
12
        self.client_socket.connect((self.host, self.port))
13
        print(f"[客户端] 已连接到 {self.host}:{self.port}")
14

15
    def send(self, message):
16
        self.client_socket.sendall(message.encode('utf-8'))
17
        response = self.client_socket.recv(4096)
18
        return response.decode('utf-8')
19

20
    def close(self):
21
        if self.client_socket:
22
            self.client_socket.close()
23
            print("[客户端] 已断开")
24

25
if __name__ == '__main__':
26
    client = TCPClient()
27
    client.connect()
28
    try:
29
        while True:
30
            msg = input("输入消息 (q 退出): ")
31
            if msg == 'q':
32
                break
33
            response = client.send(msg)
34
            print(f"服务器回复: {response}")
35
    finally:
36
        client.close()

2.3 UDP 编程实现#

1
import socket
2

3
# UDP 服务端
4
class UDPServer:
5
    def __init__(self, host='0.0.0.0', port=9999):
6
        self.host = host
7
        self.port = port
8
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
9

10
    def start(self):
11
        self.sock.bind((self.host, self.port))
12
        print(f"[UDP 服务端] 监听 {self.host}:{self.port}")
13
        try:
14
            while True:
15
                data, addr = self.sock.recvfrom(4096)
16
                message = data.decode('utf-8')
17
                print(f"[来自 {addr}] {message}")
18
                response = f"UDP 收到: {message}"
19
                self.sock.sendto(response.encode('utf-8'), addr)
20
        except KeyboardInterrupt:
21
            self.sock.close()
22
            print("[UDP 服务端] 已关闭")
23

24
# UDP 客户端
25
class UDPClient:
26
    def __init__(self, host='127.0.0.1', port=9999):
27
        self.host = host
28
        self.port = port
29
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
30

31
    def send(self, message):
32
        self.sock.sendto(message.encode('utf-8'), (self.host, self.port))
33
        data, addr = self.sock.recvfrom(4096)
34
        return data.decode('utf-8')
35

36
    def close(self):
37
        self.sock.close()
38

39
# 使用示例
40
if __name__ == '__main__':
41
    import threading
42
    import time
43

44
    server = UDPServer(port=9999)
45
    server_thread = threading.Thread(target=server.start, daemon=True)
46
    server_thread.start()
47
    time.sleep(0.5)
48

49
    client = UDPClient(port=9999)
50
    response = client.send("Hello UDP!")
51
    print(f"回复: {response}")
52
    client.close()

2.4 Socket 编程注意事项#

1
# 1. SO_REUSEADDR：端口复用，避免 TIME_WAIT 错误
2
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
3

4
# 2. TCP_NODELAY：禁用 Nagle 算法，适用于低延迟场景
5
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
6

7
# 3. 超时设置：避免 recv 永久阻塞
8
sock.settimeout(5.0)  # 5 秒超时
9

10
# 4. 发送/接收缓冲区调整
11
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)
12
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)
13

14
# 5. 非阻塞模式
15
sock.setblocking(False)
16

17
# 6. 大文件传输的分包策略
18
def send_large_file(sock, filepath):
19
    import os
20
    file_size = os.path.getsize(filepath)
21
    sock.sendall(str(file_size).encode() + b'\n')
22

23
    with open(filepath, 'rb') as f:
24
        while True:
25
            chunk = f.read(65536)  # 64KB 分片
26
            if not chunk:
27
                break
28
            sock.sendall(chunk)
29

30
def recv_all(sock, size):
31
    data = b''
32
    while len(data) < size:
33
        chunk = sock.recv(size - len(data))
34
        if not chunk:
35
            raise ConnectionError("连接中断")
36
        data += chunk
37
    return data

三、HTTP 客户端编程#

3.1 urllib 标准库#

1
import urllib.request
2
import urllib.parse
3
import urllib.error
4
import json
5

6
# 基础 GET 请求
7
url = "https://api.github.com/repos/python/cpython"
8
req = urllib.request.Request(
9
    url,
10
    headers={'User-Agent': 'Python-Network-Guide/1.0'}
11
)
12

13
try:
14
    with urllib.request.urlopen(req, timeout=10) as response:
15
        data = json.loads(response.read().decode('utf-8'))
16
        print(f"状态码: {response.status}")
17
        print(f"项目: {data['full_name']}")
18
        print(f"Stars: {data['stargazers_count']}")
19
except urllib.error.HTTPError as e:
20
    print(f"HTTP 错误: {e.code} - {e.reason}")
21
except urllib.error.URLError as e:
22
    print(f"URL 错误: {e.reason}")
23

24
# 带参数的 GET 请求
25
params = urllib.parse.urlencode({
26
    'q': 'python networking',
27
    'sort': 'stars',
28
    'per_page': 5
29
})
30
search_url = f"https://api.github.com/search/repositories?{params}"
31

32
# POST 请求
33
data = json.dumps({'name': 'test'}).encode('utf-8')
34
post_req = urllib.request.Request(
35
    "https://httpbin.org/post",
36
    data=data,
37
    headers={'Content-Type': 'application/json'},
38
    method='POST'
39
)
40

41
with urllib.request.urlopen(post_req) as resp:
42
    result = json.loads(resp.read())
43
    print(f"POST 响应: {result['json']}")

3.2 requests 库（最流行）#

1
# pip install requests
2
import requests
3

4
# 基础 GET 请求
5
response = requests.get(
6
    "https://api.github.com/repos/python/cpython",
7
    params={'sort': 'stars'},
8
    headers={'User-Agent': 'MyApp/1.0'},
9
    timeout=10
10
)
11

12
# 自动解析 JSON
13
data = response.json()
14
print(f"状态: {response.status_code}")
15
print(f"编码: {response.encoding}")
16
print(f"耗时: {response.elapsed.total_seconds():.3f}s")
17

18
# 请求参数
19
print(f"URL: {response.url}")
20
print(f"Headers: {dict(response.headers)}")
21

22
# POST 请求
23
payload = {'username': 'admin', 'password': 'secret'}
24
resp = requests.post("https://httpbin.org/post", json=payload)
25

26
# Session 复用（连接池，性能优势）
27
session = requests.Session()
28
session.headers.update({'User-Agent': 'MyApp/1.0'})
29

30
# Cookie 自动管理
31
session.get("https://httpbin.org/cookies/set", params={'name': 'value'})
32
resp = session.get("https://httpbin.org/cookies")
33
print(f"Cookies: {resp.json()}")
34

35
# 文件上传
36
with open("config.json", 'rb') as f:
37
    resp = session.post("https://httpbin.org/post", files={'file': f})
38

39
# 文件下载（流式）
40
with session.get("https://example.com/large-file.zip", stream=True) as resp:
41
    resp.raise_for_status()
42
    with open("downloaded.zip", 'wb') as f:
43
        for chunk in resp.iter_content(chunk_size=8192):
44
            f.write(chunk)
45

46
# 错误处理
47
try:
48
    resp = session.get("https://api.example.com", timeout=5)
49
    resp.raise_for_status()  # 4xx/5xx 抛出异常
50
except requests.exceptions.Timeout:
51
    print("请求超时")
52
except requests.exceptions.ConnectionError:
53
    print("连接错误")
54
except requests.exceptions.HTTPError as e:
55
    print(f"HTTP 错误: {e}")

3.3 httpx 库（现代替代品）#

1
# pip install httpx
2
import httpx
3

4
# httpx 优势：
5
# 1. 原生支持异步
6
# 2. 支持 HTTP/2
7
# 3. API 与 requests 高度兼容
8
# 4. 自动跟随重定向（可配置）
9
# 5. 支持 WebSocket
10

11
# 同步使用（类似 requests）
12
with httpx.Client(timeout=10, follow_redirects=True) as client:
13
    response = client.get(
14
        "https://api.github.com/repos/python/cpython",
15
        headers={'User-Agent': 'MyApp/1.0'}
16
    )
17
    print(f"状态: {response.status_code}")
18
    data = response.json()
19

20
# 异步使用（高性能）
21
import asyncio
22

23
async def fetch_multiple():
24
    urls = [
25
        "https://api.github.com/repos/python/cpython",
26
        "https://api.github.com/repos/python/python-docs-samples",
27
        "https://api.github.com/repos/python/peps",
28
    ]
29

30
    async with httpx.AsyncClient(timeout=10) as client:
31
        tasks = [client.get(url) for url in urls]
32
        responses = await asyncio.gather(*tasks)
33

34
        for url, resp in zip(urls, responses):
35
            data = resp.json()
36
            print(f"{url.split('/')[-1]}: {data.get('stargazers_count', 'N/A')} stars")
37

38
# HTTP/2 支持
39
with httpx.Client(http2=True) as client:
40
    resp = client.get("https://example.com")
41

42
# 重试策略
43
transport = httpx.HTTPTransport(retries=3)
44
with httpx.Client(transport=transport) as client:
45
    resp = client.get("https://api.example.com")

3.4 三大库对比#

1
特性              urllib        requests       httpx
2
─────────────────────────────────────────────────────────
3
异步支持          ❌            ❌             ✅
4
HTTP/2            ❌            ❌             ✅
5
WebSocket         ❌            ❌             ✅
6
连接池            基础          ✅             ✅
7
自动 JSON 解析    需 json.loads  ✅             ✅
8
Session 管理      手动          ✅             ✅
9
流式上传/下载     基础          ✅             ✅
10
超时/重试         基础          需配置         原生支持
11
API 兼容性        独立 API      经典 API      兼容 requests
12
推荐场景          简单脚本      通用开发       现代/异步项目

四、异步网络编程#

4.1 asyncio 基础#

1
import asyncio
2
import time
3

4
# 协程基础
5
async def hello(name, delay=1):
6
    print(f"[{time.strftime('%H:%M:%S')}] 你好, {name}!")
7
    await asyncio.sleep(delay)
8
    print(f"[{time.strftime('%H:%M:%S')}] {name} 再见!")
9
    return f"Hello {name}"
10

11
# 单协程运行
12
async def main_single():
13
    result = await hello("Alice", 1)
14
    print(f"结果: {result}")
15

16
# 并发执行多个协程
17
async def main_parallel():
18
    start = time.time()
19
    # gather 并发执行多个协程
20
    results = await asyncio.gather(
21
        hello("Alice", 2),
22
        hello("Bob", 1),
23
        hello("Charlie", 3),
24
    )
25
    elapsed = time.time() - start
26
    print(f"耗时: {elapsed:.2f}s (串行需 6s)")
27
    print(f"结果: {results}")
28

29
# 超时控制
30
async def with_timeout():
31
    try:
32
        result = await asyncio.wait_for(
33
            hello("Slow", 10),
34
            timeout=3
35
        )
36
    except asyncio.TimeoutError:
37
        print("操作超时!")
38

39
# 运行入口
40
if __name__ == '__main__':
41
    asyncio.run(main_parallel())

4.2 aiohttp 异步 HTTP 客户端#

1
# pip install aiohttp
2
import aiohttp
3
import asyncio
4
import time
5

6
async def fetch_url(session, url):
7
    """异步获取单个 URL"""
8
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
9
        return url, response.status, await response.text()
10

11
async def batch_fetch(urls):
12
    """批量异步获取"""
13
    async with aiohttp.ClientSession() as session:
14
        tasks = [fetch_url(session, url) for url in urls]
15
        results = await asyncio.gather(*tasks, return_exceptions=True)
16

17
        for url, status, body in results:
18
            if isinstance(status, Exception):
19
                print(f"{url}: 错误 - {status}")
20
            else:
21
                print(f"{url}: {status} ({len(body)} bytes)")
22

23
async def with_semaphore(urls, max_concurrent=5):
24
    """限制并发数"""
25
    sem = asyncio.Semaphore(max_concurrent)
26

27
    async def fetch_with_sem(session, url):
28
        async with sem:
29
            return await fetch_url(session, url)
30

31
    async with aiohttp.ClientSession() as session:
32
        tasks = [fetch_with_sem(session, url) for url in urls]
33
        return await asyncio.gather(*tasks)
34

35
# 高并发示例
36
urls = [
37
    f"https://httpbin.org/delay/{i % 5}" for i in range(20)
38
]
39

40
start = time.time()
41
asyncio.run(batch_fetch(urls))
42
print(f"20 个请求并发耗时: {time.time() - start:.2f}s")
43

44
# POST 请求
45
async def post_json(url, data):
46
    async with aiohttp.ClientSession() as session:
47
        async with session.post(url, json=data) as resp:
48
            return await resp.json()
49

50
# 文件上传
51
async def upload_file(url, filepath):
52
    async with aiohttp.ClientSession() as session:
53
        data = aiohttp.FormData()
54
        data.add_field('file',
55
                       open(filepath, 'rb'),
56
                       filename='data.csv',
57
                       content_type='text/csv')
58
        async with session.post(url, data=data) as resp:
59
            return await resp.json()

4.3 异步 vs 同步性能对比#

1
import requests
2
import aiohttp
3
import asyncio
4
import time
5

6
URLS = [f"https://httpbin.org/delay/1" for _ in range(20)]
7

8
# 同步版本
9
def sync_version():
10
    start = time.time()
11
    results = []
12
    for url in URLS:
13
        resp = requests.get(url, timeout=30)
14
        results.append(resp.status_code)
15
    elapsed = time.time() - start
16
    print(f"[同步] 20 请求: {elapsed:.2f}s")
17
    return elapsed
18

19
# 异步版本
20
async def async_version():
21
    start = time.time()
22
    async with aiohttp.ClientSession() as session:
23
        tasks = [session.get(url) for url in URLS]
24
        responses = await asyncio.gather(*tasks)
25
        results = [resp.status for resp in responses]
26
    elapsed = time.time() - start
27
    print(f"[异步] 20 请求: {elapsed:.2f}s")
28
    return elapsed
29

30
# 实际测试结果
31
# [同步] 20 请求: 20.45s  (每个请求约 1s，串行)
32
# [异步] 20 请求:  1.23s  (所有请求并发，几乎是 1 个请求的时间)
33
# 性能提升: 约 16.6 倍

4.4 aiohttp 服务端#

1
from aiohttp import web
2
import asyncio
3

4
# 简单的异步 Web 服务
5
async def handle_index(request):
6
    return web.json_response({
7
        'message': 'Hello, aiohttp!',
8
        'timestamp': asyncio.get_event_loop().time()
9
    })
10

11
async def handle_echo(request):
12
    name = request.match_info.get('name', 'World')
13
    return web.json_response({'echo': f'Hello, {name}!'})
14

15
async def handle_post(request):
16
    data = await request.json()
17
    return web.json_response({
18
        'received': data,
19
        'status': 'ok'
20
    })
21

22
# 中间件示例
23
async def auth_middleware(app, handler):
24
    async def middleware(request):
25
        token = request.headers.get('Authorization')
26
        if not token and request.path.startswith('/api/'):
27
            return web.json_response({'error': 'Unauthorized'}, status=401)
28
        return await handler(request)
29
    return middleware
30

31
def create_app():
32
    app = web.Application(middlewares=[auth_middleware])
33
    app.router.add_get('/', handle_index)
34
    app.router.add_get('/echo/{name}', handle_echo)
35
    app.router.add_post('/api/data', handle_post)
36
    return app
37

38
if __name__ == '__main__':
39
    app = create_app()
40
    web.run_app(app, host='0.0.0.0', port=8080)

五、WebSocket 实时通信#

5.1 WebSocket 基础#

1
# WebSocket 协议：
2
# 1. 基于 HTTP 协议升级（Upgrade）
3
# 2. 全双工通信（客户端和服务端都可主动发送）
4
# 3. 持久连接（无需每次建立 HTTP 连接）
5
# 4. 低延迟、低开销（无 HTTP 头部重复传输）
6

7
# 应用场景：
8
# - 实时聊天应用
9
# - 股票行情推送
10
# - 在线协同编辑
11
# - 游戏实时通信
12
# - IoT 设备监控
13
# - 实时通知系统

5.2 websockets 库实现#

1
# pip install websockets
2
import asyncio
3
import websockets
4

5
# WebSocket 服务端
6
async def echo_server(websocket):
7
    async for message in websocket:
8
        print(f"[服务端] 收到: {message}")
9
        response = f"服务器回复: {message}"
10
        await websocket.send(response)
11

12
async def start_server():
13
    async with websockets.serve(echo_server, "0.0.0.0", 8765):
14
        print("[WebSocket 服务端] 启动于 ws://localhost:8765")
15
        await asyncio.Future()  # 永久运行
16

17
# WebSocket 客户端
18
async def echo_client():
19
    uri = "ws://localhost:8765"
20
    async with websockets.connect(uri) as websocket:
21
        messages = ["你好", "测试消息", "WebSocket 真好用"]
22
        for msg in messages:
23
            await websocket.send(msg)
24
            response = await websocket.recv()
25
            print(f"[客户端] 发送: {msg} → 收到: {response}")
26

27
# 同时启动服务端和客户端测试
28
async def test():
29
    server_task = asyncio.create_task(start_server())
30
    await asyncio.sleep(0.5)  # 等待服务启动
31
    client_task = asyncio.create_task(echo_client())
32
    await client_task
33
    # 保持服务运行
34
    await server_task
35

36
# asyncio.run(test())

5.3 聊天室实战#

1
import asyncio
2
import websockets
3
import json
4
from collections import defaultdict
5

6
class ChatServer:
7
    def __init__(self):
8
        self.rooms = defaultdict(set)  # room_name -> set of websockets
9
        self.user_names = {}  # websocket -> username
10

11
    async def handler(self, websocket):
12
        async for message in websocket:
13
            data = json.loads(message)
14
            action = data.get('action')
15

16
            if action == 'join':
17
                await self.handle_join(websocket, data)
18
            elif action == 'leave':
19
                await self.handle_leave(websocket, data)
20
            elif action == 'message':
21
                await self.handle_message(websocket, data)
22

23
    async def handle_join(self, ws, data):
24
        room = data['room']
25
        username = data['username']
26
        self.rooms[room].add(ws)
27
        self.user_names[ws] = username
28

29
        # 通知房间内其他人
30
        notify = json.dumps({
31
            'type': 'system',
32
            'message': f'{username} 加入房间 {room}'
33
        })
34
        await self.broadcast(room, notify, exclude=ws)
35

36
        # 确认加入
37
        await ws.send(json.dumps({
38
            'type': 'joined',
39
            'room': room,
40
            'users': [self.user_names[w] for w in self.rooms[room]]
41
        }))
42

43
    async def handle_leave(self, ws, data):
44
        room = data['room']
45
        username = self.user_names.get(ws, '匿名')
46
        self.rooms[room].discard(ws)
47
        self.user_names.pop(ws, None)
48

49
        notify = json.dumps({
50
            'type': 'system',
51
            'message': f'{username} 离开房间 {room}'
52
        })
53
        await self.broadcast(room, notify)
54

55
    async def handle_message(self, ws, data):
56
        room = data['room']
57
        username = self.user_names.get(ws, '匿名')
58
        message = data['message']
59

60
        broadcast_msg = json.dumps({
61
            'type': 'message',
62
            'username': username,
63
            'message': message,
64
            'room': room
65
        })
66
        await self.broadcast(room, broadcast_msg)
67

68
    async def broadcast(self, room, message, exclude=None):
69
        for ws in self.rooms.get(room, set()):
70
            if ws != exclude:
71
                await ws.send(message)
72

73
# 启动服务器
74
if __name__ == '__main__':
75
    server = ChatServer()
76
    start = websockets.serve(server.handler, '0.0.0.0', 8765)
77
    asyncio.get_event_loop().run_until_complete(start)
78
    print("[聊天室] 启动于 ws://localhost:8765")
79
    asyncio.get_event_loop().run_forever()
80

81
# 客户端 HTML/JS 示例：
82
# const ws = new WebSocket('ws://localhost:8765');
83
# ws.send(JSON.stringify({action: 'join', room: 'general', username: 'Alice'}));
84
# ws.onmessage = (e) => console.log(JSON.parse(e.data));

六、实战：高效网络请求工具#

6.1 批量 URL 健康检查工具#

1
import asyncio
2
import time
3
from typing import List, Dict, Optional
4
import aiohttp
5

6
class URLHealthChecker:
7
    def __init__(
8
        self,
9
        timeout: float = 10.0,
10
        max_concurrent: int = 10,
11
        retry_count: int = 2
12
    ):
13
        self.timeout = timeout
14
        self.max_concurrent = max_concurrent
15
        self.retry_count = retry_count
16
        self.semaphore = asyncio.Semaphore(max_concurrent)
17

18
    async def check_url(
19
        self,
20
        session: aiohttp.ClientSession,
21
        url: str
22
    ) -> Dict:
23
        """检查单个 URL 健康状态"""
24
        async with self.semaphore:
25
            result = {
26
                'url': url,
27
                'status': None,
28
                'response_time': None,
29
                'error': None,
30
                'ok': False
31
            }
32

33
            for attempt in range(self.retry_count + 1):
34
                try:
35
                    start = time.time()
36
                    async with session.get(
37
                        url,
38
                        timeout=aiohttp.ClientTimeout(total=self.timeout),
39
                        allow_redirects=True
40
                    ) as response:
41
                        elapsed = time.time() - start
42
                        result['status'] = response.status
43
                        result['response_time'] = round(elapsed, 3)
44
                        result['ok'] = 200 <= response.status < 400
45
                        result['final_url'] = str(response.url)
46
                    break
47
                except asyncio.TimeoutError:
48
                    result['error'] = 'Timeout'
49
                    if attempt < self.retry_count:
50
                        await asyncio.sleep(1)
51
                except aiohttp.ClientError as e:
52
                    result['error'] = str(e)
53
                    if attempt < self.retry_count:
54
                        await asyncio.sleep(1)
55
                except Exception as e:
56
                    result['error'] = str(e)
57
                    break
58

59
            return result
60

61
    async def check_all(self, urls: List[str]) -> List[Dict]:
62
        """批量检查 URL"""
63
        connector = aiohttp.TCPConnector(
64
            limit=self.max_concurrent,
65
            ssl=False  # 生产环境建议开启
66
        )
67

68
        async with aiohttp.ClientSession(connector=connector) as session:
69
            tasks = [self.check_url(session, url) for url in urls]
70
            results = await asyncio.gather(*tasks, return_exceptions=True)
71

72
            # 过滤异常
73
            clean_results = []
74
            for r, url in zip(results, urls):
75
                if isinstance(r, Exception):
76
                    clean_results.append({
77
                        'url': url,
78
                        'status': None,
79
                        'response_time': None,
80
                        'error': str(r),
81
                        'ok': False
82
                    })
83
                else:
84
                    clean_results.append(r)
85

86
            return clean_results
87

88
    def run(self, urls: List[str]) -> List[Dict]:
89
        """同步入口"""
90
        return asyncio.run(self.check_all(urls))
91

92
# 使用示例
93
if __name__ == '__main__':
94
    urls = [
95
        "https://www.google.com",
96
        "https://www.github.com",
97
        "https://www.python.org",
98
        "https://www.cloudflare.com",
99
        "https://example.com",
100
        "https://httpbin.org/status/200",
101
        "https://httpbin.org/status/404",
102
        "https://invalid-domain-that-does-not-exist.com",
103
    ]
104

105
    checker = URLHealthChecker(timeout=5, max_concurrent=5)
106

107
    start = time.time()
108
    results = checker.run(urls)
109
    total_time = time.time() - start
110

111
    # 输出报告
112
    print(f"\n{'='*60}")
113
    print(f"URL 健康检查报告 - 耗时 {total_time:.2f}s")
114
    print(f"{'='*60}")
115

116
    ok_count = sum(1 for r in results if r['ok'])
117
    fail_count = len(results) - ok_count
118

119
    for r in results:
120
        status_icon = "✅" if r['ok'] else "❌"
121
        rt = f"{r['response_time']}s" if r['response_time'] else "N/A"
122
        err = f" ({r['error']})" if r['error'] else ""
123
        print(f"{status_icon} [{r['status'] or 'ERR'}] {rt:>8} {r['url']}{err}")
124

125
    print(f"\n统计: {ok_count} 通过, {fail_count} 失败, 共 {len(results)} 个 URL")

6.2 使用示例#

1
输出示例：
2
============================================================
3
URL 健康检查报告 - 耗时 3.45s
4
============================================================
5
✅ [200]    0.312s https://www.google.com
6
✅ [200]    0.523s https://www.github.com
7
✅ [200]    0.891s https://www.python.org
8
✅ [200]    0.234s https://www.cloudflare.com
9
✅ [200]    0.015s https://example.com
10
✅ [200]    0.256s https://httpbin.org/status/200
11
❌ [404]    0.198s https://httpbin.org/status/404
12
❌ [ERR]      N/A https://invalid-domain-that-does-not-exist.com (Cannot resolve host)
13

14
统计: 6 通过, 2 失败, 共 8 个 URL

七、最佳实践与总结#

7.1 库选择决策树#

1
需要网络编程？
2
├── 底层自定义协议？
3
│   └── Python socket 模块
4
├── HTTP/REST API 调用？
5
│   ├── 简单脚本 → requests
6
│   ├── 需要异步/HTTP/2 → httpx
7
│   └── 高并发爬虫 → aiohttp
8
├── 实时双向通信？
9
│   └── websockets
10
├── Web 服务开发？
11
│   ├── 简单 API → aiohttp / Flask
12
│   ├── 高性能 → FastAPI
13
│   └── 微服务 → gRPC + Protobuf
14
└── 其他需求？
15
    ├── 邮件 → smtplib
16
    ├── FTP → ftplib
17
    └── DNS → dnspython

7.2 性能优化要点#

1
# 1. 连接复用：使用 Session/ClientSession
2
session = requests.Session()  # 复用 TCP 连接
3
aiohttp_session = aiohttp.ClientSession()  # 异步复用
4

5
# 2. 连接池配置
6
from requests.adapters import HTTPAdapter
7
from urllib3.util.retry import Retry
8

9
adapter = HTTPAdapter(
10
    pool_connections=100,
11
    pool_maxsize=100,
12
    max_retries=Retry(total=3, backoff_factor=1)
13
)
14
session.mount('http://', adapter)
15
session.mount('https://', adapter)
16

17
# 3. 批量并发：异步 + Semaphore 控制并发数
18
sem = asyncio.Semaphore(10)  # 限制 10 个并发
19

20
# 4. 超时设置：永远不要省略
21
# connect timeout + read timeout
22
resp = requests.get(url, timeout=(5, 30))
23

24
# 5. 流式处理：大文件不要一次性读入内存
25
for chunk in response.iter_content(chunk_size=8192):
26
    file.write(chunk)
27

28
# 6. 合理的 User-Agent：避免被封禁
29
headers = {
30
    'User-Agent': 'Mozilla/5.0 (compatible; MyBot/1.0)'
31
}
32

33
# 7. 错误处理：重试 + 降级
34
try:
35
    resp = session.get(url, timeout=10)
36
    resp.raise_for_status()
37
except requests.exceptions.RequestException:
38
    # 降级：使用备份 URL 或缓存
39
    pass

7.3 常见问题排查#

1
# 问题1: 连接超时/SSL 错误
2
# 解决方案：检查网络、配置代理、验证证书
3
import requests
4
# 跳过 SSL 验证（仅开发环境）
5
resp = requests.get(url, verify=False)
6

7
# 问题2: 连接过多导致端口耗尽
8
# 解决方案：使用 Session 复用、控制连接数
9
# 错误：每次请求创建新连接
10
for url in urls:
11
    requests.get(url)  # 新连接
12
# 正确：Session 复用
13
with requests.Session() as s:
14
    for url in urls:
15
        s.get(url)  # 复用连接
16

17
# 问题3: aiohttp 与 requests 混用导致事件循环问题
18
# 解决方案：异步进程完全使用 aiohttp
19
# 错误：在 async 函数内使用 requests
20
async def bad():
21
    requests.get(url)  # 阻塞事件循环！
22
# 正确：完全使用异步库
23
async def good():
24
    async with session.get(url) as resp:
25
        return await resp.json()
26

27
# 问题4: WebSocket 频繁断连
28
# 解决方案：心跳机制 + 自动重连
29
async def websocket_with_heartbeat(uri):
30
    while True:
31
        try:
32
            async with websockets.connect(
33
                uri,
34
                ping_interval=30,
35
                ping_timeout=10
36
            ) as ws:
37
                async for message in ws:
38
                    handle_message(message)
39
        except websockets.ConnectionClosed:
40
            print("连接断开，3 秒后重连...")
41
            await asyncio.sleep(3)

7.4 学习资源推荐#

1
1. 官方文档
2
   - Python socket: https://docs.python.org/3/library/socket.html
3
   - requests: https://requests.readthedocs.io/
4
   - httpx: https://www.python-httpx.org/
5
   - aiohttp: https://docs.aiohttp.org/
6
   - websockets: https://websockets.readthedocs.io/
7

8
2. 进阶书籍
9
   - 《Python 网络编程》
10
   - 《O'Reilly Python Cookbook》
11

12
3. 实战项目
13
   - 构建个人代理工具
14
   - 开发小型 Web 框架
15
   - 编写网络爬虫/API 客户端
16
   - 实现实时数据推送服务

总结#

Python 网络编程体系庞大，但可以按层次循序渐进地学习：

基础层：理解 OSI/TCP-IP 模型、TCP/UDP 协议、Socket 编程
应用层：掌握 requests/httpx 等 HTTP 客户端库
进阶层：学习 asyncio + aiohttp 异步编程应对高并发场景
高级层：WebSocket 实时通信、性能优化、错误处理

实践是最好的学习方式。建议从简单的 HTTP 请求开始，逐步深入到异步编程和实时通信，构建属于自己的网络工具集。