第11章网络编程¶

网络编程是现代软件开发中不可或缺的技能，Python提供了丰富的网络编程库和工具，让开发者能够轻松构建各种网络应用。本章将从基础的Socket编程开始，逐步深入到HTTP客户端和服务器开发、异步网络编程以及网络安全等高级主题。

本系列文章所使用到的示例源码：Python从入门到精通示例代码

11.1 网络编程基础¶

网络协议概述¶

网络协议是计算机网络中进行数据交换而建立的规则、标准或约定。理解这些协议对于网络编程至关重要。

TCP/IP协议栈¶

TCP/IP协议栈是互联网的基础，它包含四个层次：

应用层：HTTP、HTTPS、FTP、SMTP、DNS等
传输层：TCP、UDP
网络层：IP、ICMP、ARP
链路层：以太网、WiFi等

常用网络协议¶

HTTP/HTTPS：超文本传输协议，Web应用的基础
FTP：文件传输协议
SMTP：简单邮件传输协议
DNS：域名系统
SSH：安全外壳协议

客户端-服务器模型¶

客户端-服务器（C/S）架构是网络应用的基本模式：

客户端：发起请求的一方
服务器：响应请求的一方
请求-响应模式：客户端发送请求，服务器处理并返回响应

IP地址和端口¶

IP地址¶

IPv4：32位地址，如 192.168.1.1
IPv6：128位地址，如 2001:db8::1

端口号¶

端口号用于标识主机上的不同服务：

知名端口（0-1023）：HTTP(80)、HTTPS(443)、FTP(21)、SSH(22)
注册端口（1024-49151）：应用程序注册使用
动态端口（49152-65535）：临时分配使用

11.2 Socket编程¶

Socket（套接字）是网络编程的基础，它提供了进程间通信的接口。

Socket概念¶

Socket是网络通信的端点，可以理解为网络编程的”插座”。每个Socket都有一个地址，包括IP地址和端口号。

套接字类型¶

TCP Socket（SOCK_STREAM）：面向连接，可靠传输
UDP Socket（SOCK_DGRAM）：无连接，不可靠但高效

Python的socket模块¶

Python的socket模块提供了底层网络接口：

import socket

# 创建TCP socket
tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 创建UDP socket
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

地址族¶

AF_INET：IPv4地址族
AF_INET6：IPv6地址族
AF_UNIX：Unix域套接字

TCP Socket编程实战¶

让我们通过实际的代码示例来学习TCP Socket编程。

TCP服务器端¶

首先创建一个TCP服务器，它能够处理多个客户端连接：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TCP服务器示例
"""

import socket
import threading
import time

def handle_client(client_socket, client_address):
    """处理客户端连接"""
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端 {client_address} 已连接")

    try:
        while True:
            # 接收客户端数据
            data = client_socket.recv(1024)
            if not data:
                break

            message = data.decode('utf-8')
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 收到来自 {client_address} 的消息: {message}")

            # 回复客户端
            response = f"服务器收到: {message}"
            client_socket.send(response.encode('utf-8'))

    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 处理客户端 {client_address} 时发生错误: {e}")
    finally:
        client_socket.close()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端 {client_address} 已断开连接")

def start_server():
    """启动TCP服务器"""
    # 创建socket对象
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    # 设置socket选项，允许地址重用
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    # 绑定地址和端口
    host = 'localhost'
    port = 8888
    server_socket.bind((host, port))

    # 开始监听
    server_socket.listen(5)
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] TCP服务器启动，监听 {host}:{port}")

    try:
        while True:
            # 接受客户端连接
            client_socket, client_address = server_socket.accept()

            # 为每个客户端创建新线程
            client_thread = threading.Thread(
                target=handle_client,
                args=(client_socket, client_address)
            )
            client_thread.daemon = True
            client_thread.start()

    except KeyboardInterrupt:
        print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服务器正在关闭...")
    finally:
        server_socket.close()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服务器已关闭")

if __name__ == '__main__':
    start_server()

TCP客户端¶

接下来创建TCP客户端来连接服务器：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TCP客户端示例
"""

import socket
import time

def start_client():
    """启动TCP客户端"""
    # 创建socket对象
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    try:
        # 连接服务器
        host = 'localhost'
        port = 8888
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 正在连接服务器 {host}:{port}...")
        client_socket.connect((host, port))
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 已连接到服务器")

        # 发送消息
        messages = [
            "Hello, Server!",
            "这是第二条消息",
            "Python网络编程测试",
            "再见!"
        ]

        for message in messages:
            # 发送数据
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送消息: {message}")
            client_socket.send(message.encode('utf-8'))

            # 接收服务器响应
            response = client_socket.recv(1024)
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服务器响应: {response.decode('utf-8')}")

            time.sleep(1)  # 等待1秒

    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端错误: {e}")
    finally:
        client_socket.close()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端已断开连接")

if __name__ == '__main__':
    start_client()

运行TCP示例¶

首先启动服务器：

python tcp_server.py

服务器输出：

[2025-08-15 21:47:55] TCP服务器启动，监听 localhost:8888

然后在另一个终端启动客户端：

python tcp_client.py

客户端输出：

[2025-08-15 21:48:07] 正在连接服务器 localhost:8888...
[2025-08-15 21:48:07] 已连接到服务器
[2025-08-15 21:48:07] 发送消息: Hello, Server!
[2025-08-15 21:48:07] 服务器响应: 服务器收到: Hello, Server!
[2025-08-15 21:48:08] 发送消息: 这是第二条消息
[2025-08-15 21:48:08] 服务器响应: 服务器收到: 这是第二条消息
[2025-08-15 21:48:09] 发送消息: Python网络编程测试
[2025-08-15 21:48:09] 服务器响应: 服务器收到: Python网络编程测试
[2025-08-15 21:48:10] 发送消息: 再见!
[2025-08-15 21:48:10] 服务器响应: 服务器收到: 再见!
[2025-08-15 21:48:11] 客户端已断开连接

服务器端同时显示：

[2025-08-15 21:48:07] 客户端 ('127.0.0.1', 50531) 已连接
[2025-08-15 21:48:07] 收到来自 ('127.0.0.1', 50531) 的消息: Hello, Server!
[2025-08-15 21:48:08] 收到来自 ('127.0.0.1', 50531) 的消息: 这是第二条消息
[2025-08-15 21:48:09] 收到来自 ('127.0.0.1', 50531) 的消息: Python网络编程测试
[2025-08-15 21:48:10] 收到来自 ('127.0.0.1', 50531) 的消息: 再见!
[2025-08-15 21:48:11] 客户端 ('127.0.0.1', 50531) 已断开连接

UDP Socket编程实战¶

UDP是无连接的协议，不需要建立连接就可以发送数据，适合对实时性要求高的应用。

UDP服务器端¶

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
UDP服务器示例
"""

import socket
import time

def start_udp_server():
    """启动UDP服务器"""
    # 创建UDP socket对象
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    # 绑定地址和端口
    host = 'localhost'
    port = 9999
    server_socket.bind((host, port))

    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP服务器启动，监听 {host}:{port}")

    try:
        while True:
            # 接收数据和客户端地址
            data, client_address = server_socket.recvfrom(1024)
            message = data.decode('utf-8')

            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 收到来自 {client_address} 的消息: {message}")

            # 回复客户端
            response = f"UDP服务器收到: {message}"
            server_socket.sendto(response.encode('utf-8'), client_address)
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 已回复客户端 {client_address}")

    except KeyboardInterrupt:
        print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP服务器正在关闭...")
    finally:
        server_socket.close()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP服务器已关闭")

if __name__ == '__main__':
    start_udp_server()

UDP客户端¶

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
UDP客户端示例
"""

import socket
import time

def start_udp_client():
    """启动UDP客户端"""
    # 创建UDP socket对象
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    try:
        # 服务器地址
        server_host = 'localhost'
        server_port = 9999
        server_address = (server_host, server_port)

        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP客户端启动，目标服务器 {server_host}:{server_port}")

        # 发送消息列表
        messages = [
            "Hello UDP Server!",
            "这是UDP通信测试",
            "无连接协议演示",
            "UDP消息结束"
        ]

        for message in messages:
            # 发送数据到服务器
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送消息: {message}")
            client_socket.sendto(message.encode('utf-8'), server_address)

            # 接收服务器响应
            response, server_addr = client_socket.recvfrom(1024)
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服务器 {server_addr} 响应: {response.decode('utf-8')}")

            time.sleep(1)  # 等待1秒

    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP客户端错误: {e}")
    finally:
        client_socket.close()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP客户端已关闭")

if __name__ == '__main__':
    start_udp_client()

运行UDP示例¶

启动UDP服务器：

python udp_server.py

服务器输出：

[2025-08-15 21:48:29] UDP服务器启动，监听 localhost:9999

启动UDP客户端：

python udp_client.py

客户端输出：

[2025-08-15 21:48:36] UDP客户端启动，目标服务器 localhost:9999
[2025-08-15 21:48:36] 发送消息: Hello UDP Server!
[2025-08-15 21:48:36] 服务器 ('127.0.0.1', 9999) 响应: UDP服务器收到: Hello UDP Server!
[2025-08-15 21:48:37] 发送消息: 这是UDP通信测试
[2025-08-15 21:48:37] 服务器 ('127.0.0.1', 9999) 响应: UDP服务器收到: 这是UDP通信测试
[2025-08-15 21:48:38] 发送消息: 无连接协议演示
[2025-08-15 21:48:38] 服务器 ('127.0.0.1', 9999) 响应: UDP服务器收到: 无连接协议演示
[2025-08-15 21:48:39] 发送消息: UDP消息结束
[2025-08-15 21:48:39] 服务器 ('127.0.0.1', 9999) 响应: UDP服务器收到: UDP消息结束
[2025-08-15 21:48:40] UDP客户端已关闭

服务器端显示：

[2025-08-15 21:48:36] 收到来自 ('127.0.0.1', 57540) 的消息: Hello UDP Server!
[2025-08-15 21:48:36] 已回复客户端 ('127.0.0.1', 57540)
[2025-08-15 21:48:37] 收到来自 ('127.0.0.1', 57540) 的消息: 这是UDP通信测试
[2025-08-15 21:48:37] 已回复客户端 ('127.0.0.1', 57540)
[2025-08-15 21:48:38] 收到来自 ('127.0.0.1', 57540) 的消息: 无连接协议演示
[2025-08-15 21:48:38] 已回复客户端 ('127.0.0.1', 57540)
[2025-08-15 21:48:39] 收到来自 ('127.0.0.1', 57540) 的消息: UDP消息结束
[2025-08-15 21:48:39] 已回复客户端 ('127.0.0.1', 57540)

TCP vs UDP 对比¶

特性	TCP	UDP
连接性	面向连接	无连接
可靠性	可靠传输	不可靠传输
速度	较慢	较快
数据完整性	保证	不保证
适用场景	文件传输、网页浏览	视频直播、游戏

Socket编程最佳实践¶

1. 异常处理¶

try:
    # socket操作
    pass
except socket.error as e:
    print(f"Socket错误: {e}")
except Exception as e:
    print(f"其他错误: {e}")
finally:
    # 确保关闭socket
    if 'sock' in locals():
        sock.close()

2. 超时设置¶

# 设置超时时间
sock.settimeout(10.0)  # 10秒超时

3. 地址重用¶

# 允许地址重用，避免"Address already in use"错误
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

4. 缓冲区管理¶

# 设置接收缓冲区大小
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)

# 设置发送缓冲区大小
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)

11.3 HTTP客户端编程¶

HTTP（超文本传输协议）是Web应用的基础协议。Python提供了多种方式来进行HTTP客户端编程。

HTTP协议基础¶

HTTP请求方法¶

GET：获取资源
POST：提交数据
PUT：更新资源
DELETE：删除资源
HEAD：获取响应头
OPTIONS：获取支持的方法

HTTP状态码¶

2xx：成功（200 OK、201 Created、204 No Content）
3xx：重定向（301 Moved Permanently、302 Found、304 Not Modified）
4xx：客户端错误（400 Bad Request、401 Unauthorized、404 Not Found）
5xx：服务器错误（500 Internal Server Error、502 Bad Gateway、503 Service Unavailable）

urllib模块¶

urllib是Python标准库中的HTTP客户端模块，无需安装第三方库。

urllib.request基本使用¶

import urllib.request
import urllib.parse
import json

# GET请求
response = urllib.request.urlopen('https://httpbin.org/get')
print(f"状态码: {response.getcode()}")
print(f"响应头: {dict(response.headers)}")
data = response.read().decode('utf-8')
print(f"响应内容: {data}")

# POST请求
post_data = {'key': 'value', 'name': 'Python'}
data = urllib.parse.urlencode(post_data).encode('utf-8')
req = urllib.request.Request('https://httpbin.org/post', data=data)
response = urllib.request.urlopen(req)
print(f"POST响应: {response.read().decode('utf-8')}")

requests库¶

requests库是Python中最流行的HTTP客户端库，提供了更简洁的API。

安装requests¶

pip install requests

requests基本使用¶

让我们通过实际的代码示例来学习HTTP客户端编程：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
HTTP客户端编程示例
"""

import urllib.request
import urllib.parse
import urllib.error
import requests
import json
import time

def test_urllib():
    """测试urllib模块"""
    print("=== urllib模块测试 ===")

    try:
        # GET请求
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送urllib GET请求...")
        response = urllib.request.urlopen('https://httpbin.org/get', timeout=5)
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] GET请求状态码: {response.getcode()}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 响应头Content-Type: {response.headers.get('Content-Type')}")

        data = json.loads(response.read().decode('utf-8'))
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端IP: {data.get('origin')}")

    except urllib.error.URLError as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] urllib GET请求失败: {e}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] urllib GET请求异常: {e}")

    try:
        # POST请求
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送urllib POST请求...")
        post_data = {
            'name': 'Python网络编程',
            'version': '3.9',
            'framework': 'urllib'
        }

        data = urllib.parse.urlencode(post_data).encode('utf-8')
        req = urllib.request.Request('https://httpbin.org/post', data=data)
        req.add_header('Content-Type', 'application/x-www-form-urlencoded')

        response = urllib.request.urlopen(req, timeout=5)
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] POST请求状态码: {response.getcode()}")

        result = json.loads(response.read().decode('utf-8'))
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 提交的表单数据: {result.get('form')}")

    except urllib.error.URLError as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] urllib POST请求失败: {e}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] urllib POST请求异常: {e}")

def test_requests():
    """测试requests库"""
    print("\n=== requests库测试 ===")

    try:
        # GET请求
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送requests GET请求...")
        response = requests.get('https://httpbin.org/get', timeout=5)
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] GET请求状态码: {response.status_code}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 响应头Content-Type: {response.headers.get('Content-Type')}")

        data = response.json()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端IP: {data.get('origin')}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] User-Agent: {data.get('headers', {}).get('User-Agent')}")

    except requests.RequestException as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] requests GET请求失败: {e}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] requests GET请求异常: {e}")

    try:
        # POST请求
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送requests POST请求...")
        post_data = {
            'name': 'Python网络编程',
            'version': '3.9',
            'framework': 'requests',
            'features': ['简单易用', '功能强大', '社区活跃']
        }

        headers = {
            'User-Agent': 'Python-HTTP-Client/1.0',
            'Content-Type': 'application/json'
        }

        response = requests.post(
            'https://httpbin.org/post',
            json=post_data,
            headers=headers,
            timeout=5
        )

        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] POST请求状态码: {response.status_code}")

        result = response.json()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 提交的JSON数据: {result.get('json')}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 请求头User-Agent: {result.get('headers', {}).get('User-Agent')}")

    except requests.RequestException as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] requests POST请求失败: {e}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] requests POST请求异常: {e}")

if __name__ == '__main__':
    test_urllib()
    test_requests()

运行HTTP客户端示例¶

python http_client_demo.py

实际运行输出：

=== urllib模块测试 ===
[2025-08-15 21:48:48] 发送urllib GET请求...
[2025-08-15 21:48:48] urllib GET请求失败: <urlopen error timed out>
[2025-08-15 21:48:53] 发送urllib POST请求...
[2025-08-15 21:48:53] urllib POST请求失败: <urlopen error timed out>

=== requests库测试 ===
[2025-08-15 21:48:58] 发送requests GET请求...
[2025-08-15 21:48:59] GET请求状态码: 200
[2025-08-15 21:48:59] 响应头Content-Type: application/json
[2025-08-15 21:48:59] 客户端IP: 116.21.38.173
[2025-08-15 21:48:59] User-Agent: python-requests/2.32.3
[2025-08-15 21:48:59] 发送requests POST请求...
[2025-08-15 21:49:00] POST请求状态码: 200
[2025-08-15 21:49:00] 提交的JSON数据: {'name': 'Python网络编程', 'version': '3.9', 'framework': 'requests', 'features': ['简单易用', '功能强大', '社区活跃']}
[2025-08-15 21:49:00] 请求头User-Agent: Python-HTTP-Client/1.0

从输出可以看到，requests库相比urllib更加稳定和易用，成功完成了HTTP请求。

requests高级功能¶

会话管理¶

import requests

# 创建会话对象
session = requests.Session()
session.headers.update({'User-Agent': 'My-App/1.0'})

# 使用会话发送请求
response = session.get('https://httpbin.org/get')
print(response.json())

# 会话会自动处理cookies
session.get('https://httpbin.org/cookies/set/sessioncookie/123456789')
response = session.get('https://httpbin.org/cookies')
print(response.json())

文件上传¶

import requests

# 上传文件
files = {'file': open('example.txt', 'rb')}
response = requests.post('https://httpbin.org/post', files=files)
print(response.json())

超时和重试¶

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

# 配置重试策略
retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    method_whitelist=["HEAD", "GET", "OPTIONS"]
)

adapter = HTTPAdapter(max_retries=retry_strategy)
session = requests.Session()
session.mount("http://", adapter)
session.mount("https://", adapter)

# 发送请求（带超时和重试）
response = session.get('https://httpbin.org/get', timeout=5)

11.4 Web服务器开发¶

Python提供了多种方式来开发Web服务器，从简单的静态文件服务器到复杂的Web应用框架。

HTTP服务器基础¶

HTTP服务器的基本工作流程：

监听端口：绑定到指定的IP地址和端口
接受连接：等待客户端连接
解析请求：解析HTTP请求报文
处理请求：根据请求执行相应的业务逻辑
生成响应：构造HTTP响应报文
发送响应：将响应发送给客户端

http.server模块¶

Python标准库提供了http.server模块来快速创建HTTP服务器。

简单HTTP服务器¶

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
简单HTTP服务器示例
"""

import http.server
import socketserver
import json
import urllib.parse
import time
from datetime import datetime

class CustomHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
    """自定义HTTP请求处理器"""

    def do_GET(self):
        """处理GET请求"""
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] GET请求: {self.path} 来自 {self.client_address[0]}")

        if self.path == '/':
            self.send_html_response(self.get_home_page())
        elif self.path == '/api/info':
            self.send_json_response(self.get_server_info())
        elif self.path.startswith('/api/time'):
            self.send_json_response(self.get_current_time())
        else:
            self.send_error(404, "页面未找到")

    def do_POST(self):
        """处理POST请求"""
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] POST请求: {self.path} 来自 {self.client_address[0]}")

        if self.path == '/api/echo':
            # 读取请求体
            content_length = int(self.headers.get('Content-Length', 0))
            post_data = self.rfile.read(content_length)

            try:
                # 尝试解析JSON
                data = json.loads(post_data.decode('utf-8'))
                response = {
                    'status': 'success',
                    'message': '数据接收成功',
                    'received_data': data,
                    'timestamp': datetime.now().isoformat()
                }
                self.send_json_response(response)
            except json.JSONDecodeError:
                # 处理表单数据
                form_data = urllib.parse.parse_qs(post_data.decode('utf-8'))
                response = {
                    'status': 'success',
                    'message': '表单数据接收成功',
                    'form_data': form_data,
                    'timestamp': datetime.now().isoformat()
                }
                self.send_json_response(response)
        else:
            self.send_error(404, "API端点未找到")

    def send_html_response(self, html_content):
        """发送HTML响应"""
        self.send_response(200)
        self.send_header('Content-Type', 'text/html; charset=utf-8')
        self.send_header('Server', 'Python-HTTP-Server/1.0')
        self.end_headers()
        self.wfile.write(html_content.encode('utf-8'))

    def send_json_response(self, data):
        """发送JSON响应"""
        json_data = json.dumps(data, ensure_ascii=False, indent=2)
        self.send_response(200)
        self.send_header('Content-Type', 'application/json; charset=utf-8')
        self.send_header('Server', 'Python-HTTP-Server/1.0')
        self.send_header('Access-Control-Allow-Origin', '*')
        self.end_headers()
        self.wfile.write(json_data.encode('utf-8'))

    def get_home_page(self):
        """获取首页HTML"""
        return """
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Python HTTP服务器</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 40px; }
        .container { max-width: 800px; margin: 0 auto; }
        .api-list { background: #f5f5f5; padding: 20px; border-radius: 5px; }
        .api-item { margin: 10px 0; }
        .method { color: #007bff; font-weight: bold; }
    </style>
</head>
<body>
    <div class="container">
        <h1>Python HTTP服务器演示</h1>
        <p>这是一个使用Python http.server模块创建的简单Web服务器。</p>

        <h2>可用的API端点：</h2>
        <div class="api-list">
            <div class="api-item">
                <span class="method">GET</span> /api/info - 获取服务器信息
            </div>
            <div class="api-item">
                <span class="method">GET</span> /api/time - 获取当前时间
            </div>
            <div class="api-item">
                <span class="method">POST</span> /api/echo - 回显提交的数据
            </div>
        </div>

        <h2>测试示例：</h2>
        <pre>
# 获取服务器信息
curl http://localhost:8000/api/info

# 提交JSON数据
curl -X POST -H "Content-Type: application/json" \
     -d '{"name":"测试","value":123}' \
     http://localhost:8000/api/echo
        </pre>
    </div>
</body>
</html>
        """

    def get_server_info(self):
        """获取服务器信息"""
        return {
            'server': 'Python HTTP Server',
            'version': '1.0',
            'timestamp': datetime.now().isoformat(),
            'client_ip': self.client_address[0],
            'user_agent': self.headers.get('User-Agent', 'Unknown')
        }

    def get_current_time(self):
        """获取当前时间"""
        now = datetime.now()
        return {
            'current_time': now.isoformat(),
            'timestamp': now.timestamp(),
            'formatted_time': now.strftime('%Y年%m月%d日 %H:%M:%S'),
            'timezone': str(now.astimezone().tzinfo)
        }

    def log_message(self, format, *args):
        """自定义日志格式"""
        pass  # 我们已经在do_GET和do_POST中记录了日志

def start_server():
    """启动HTTP服务器"""
    host = 'localhost'
    port = 8000

    with socketserver.TCPServer((host, port), CustomHTTPRequestHandler) as httpd:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] HTTP服务器启动，监听 http://{host}:{port}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 按 Ctrl+C 停止服务器")

        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服务器正在关闭...")
        finally:
            httpd.server_close()
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服务器已关闭")

if __name__ == '__main__':
    start_server()

运行HTTP服务器示例¶

启动服务器：

python simple_http_server.py

服务器输出：

[2025-08-15 21:49:08] HTTP服务器启动，监听 http://localhost:8000
[2025-08-15 21:49:08] 按 Ctrl+C 停止服务器

测试API端点：

curl http://localhost:8000/api/info

响应结果：

{
  "server": "Python HTTP Server",
  "version": "1.0",
  "timestamp": "2025-08-15T21:49:15.123456",
  "client_ip": "127.0.0.1",
  "user_agent": "curl/8.7.1"
}

服务器日志：

[2025-08-15 21:49:15] GET请求: /api/info 来自 127.0.0.1

WSGI协议¶

WSGI（Web Server Gateway Interface）是Python Web应用和Web服务器之间的标准接口。

WSGI应用示例¶

def simple_wsgi_app(environ, start_response):
    """简单的WSGI应用"""
    status = '200 OK'
    headers = [('Content-Type', 'text/html; charset=utf-8')]
    start_response(status, headers)

    html = f"""
    <html>
    <body>
        <h1>WSGI应用示例</h1>
        <p>请求方法: {environ['REQUEST_METHOD']}</p>
        <p>请求路径: {environ['PATH_INFO']}</p>
        <p>查询字符串: {environ['QUERY_STRING']}</p>
    </body>
    </html>
    """

    return [html.encode('utf-8')]

# 使用wsgiref运行WSGI应用
from wsgiref.simple_server import make_server

if __name__ == '__main__':
    server = make_server('localhost', 8080, simple_wsgi_app)
    print("WSGI服务器启动在 http://localhost:8080")
    server.serve_forever()

11.5 网络数据处理¶

在网络编程中，数据的序列化、压缩和加密是常见的需求。

数据序列化¶

JSON序列化¶

import json

# Python对象转JSON
data = {
    'name': '张三',
    'age': 25,
    'skills': ['Python', 'JavaScript', 'SQL'],
    'is_active': True
}

# 序列化
json_str = json.dumps(data, ensure_ascii=False, indent=2)
print(f"JSON字符串: {json_str}")

# 反序列化
parsed_data = json.loads(json_str)
print(f"解析后的数据: {parsed_data}")

pickle序列化¶

import pickle

# 序列化Python对象
data = {'name': '测试', 'numbers': [1, 2, 3, 4, 5]}
serialized = pickle.dumps(data)
print(f"序列化后的字节数据长度: {len(serialized)}")

# 反序列化
deserialized = pickle.loads(serialized)
print(f"反序列化后的数据: {deserialized}")

数据压缩¶

gzip压缩¶

import gzip
import json

# 原始数据
data = {'message': '这是一个测试消息' * 100}
json_data = json.dumps(data, ensure_ascii=False)

print(f"原始数据大小: {len(json_data.encode('utf-8'))} 字节")

# 压缩数据
compressed = gzip.compress(json_data.encode('utf-8'))
print(f"压缩后大小: {len(compressed)} 字节")
print(f"压缩比: {len(compressed) / len(json_data.encode('utf-8')):.2%}")

# 解压数据
decompressed = gzip.decompress(compressed).decode('utf-8')
restored_data = json.loads(decompressed)
print(f"解压后数据正确: {restored_data == data}")

数据加密¶

hashlib哈希¶

import hashlib

# 计算哈希值
data = "Python网络编程"

# MD5哈希
md5_hash = hashlib.md5(data.encode('utf-8')).hexdigest()
print(f"MD5: {md5_hash}")

# SHA256哈希
sha256_hash = hashlib.sha256(data.encode('utf-8')).hexdigest()
print(f"SHA256: {sha256_hash}")

11.6 异步网络编程¶

异步编程可以显著提高网络应用的性能，特别是在处理大量并发连接时。

异步编程概念¶

同步vs异步¶

同步编程：代码按顺序执行，每个操作完成后才执行下一个
异步编程：可以在等待I/O操作时执行其他任务

事件循环¶

事件循环是异步编程的核心，它负责调度和执行异步任务。

asyncio模块¶

Python 3.4+引入了asyncio模块，提供了异步编程的基础设施。

基本概念¶

import asyncio

# 协程函数
async def hello_world():
    print("Hello")
    await asyncio.sleep(1)  # 异步等待
    print("World")

# 运行协程
asyncio.run(hello_world())

异步网络编程实战¶

让我们创建一个完整的异步网络编程示例：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
异步网络编程示例
"""

import asyncio
import json
import time
from datetime import datetime

class AsyncTCPServer:
    """异步TCP服务器"""

    def __init__(self, host='localhost', port=9000):
        self.host = host
        self.port = port
        self.clients = set()
        self.client_count = 0

    async def handle_client(self, reader, writer):
        """处理客户端连接"""
        client_address = writer.get_extra_info('peername')
        self.client_count += 1
        client_id = self.client_count
        self.clients.add(writer)

        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端 {client_id} ({client_address}) 已连接")

        try:
            # 发送欢迎消息
            welcome_msg = f"欢迎连接到异步服务器！您是第 {client_id} 个客户端\n"
            writer.write(welcome_msg.encode('utf-8'))
            await writer.drain()

            while True:
                # 读取客户端数据
                data = await reader.read(1024)
                if not data:
                    break

                message = data.decode('utf-8').strip()
                print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端 {client_id} 发送: {message}")

                # 处理特殊命令
                if message.lower() == 'time':
                    response = f"当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
                elif message.lower() == 'clients':
                    response = f"当前连接客户端数: {len(self.clients)}\n"
                elif message.lower().startswith('broadcast:'):
                    broadcast_msg = message[10:].strip()
                    await self.broadcast_message(f"[广播] 客户端 {client_id}: {broadcast_msg}\n", exclude=writer)
                    response = "广播消息已发送\n"
                elif message.lower() == 'quit':
                    response = "再见！\n"
                    writer.write(response.encode('utf-8'))
                    await writer.drain()
                    break
                else:
                    response = f"服务器收到: {message}\n"

                # 发送响应
                writer.write(response.encode('utf-8'))
                await writer.drain()

        except asyncio.CancelledError:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端 {client_id} 连接被取消")
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 处理客户端 {client_id} 时发生错误: {e}")
        finally:
            self.clients.discard(writer)
            writer.close()
            await writer.wait_closed()
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端 {client_id} ({client_address}) 已断开连接")

    async def broadcast_message(self, message, exclude=None):
        """向所有客户端广播消息"""
        if not self.clients:
            return

        tasks = []
        for client in self.clients.copy():
            if client != exclude and not client.is_closing():
                tasks.append(self.send_to_client(client, message))

        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

    async def send_to_client(self, writer, message):
        """向单个客户端发送消息"""
        try:
            writer.write(message.encode('utf-8'))
            await writer.drain()
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送消息到客户端失败: {e}")
            self.clients.discard(writer)

    async def start_server(self):
        """启动异步服务器"""
        server = await asyncio.start_server(
            self.handle_client,
            self.host,
            self.port
        )

        addr = server.sockets[0].getsockname()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 异步TCP服务器启动，监听 {addr[0]}:{addr[1]}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 支持的命令: time, clients, broadcast:<消息>, quit")

        async with server:
            await server.serve_forever()

class AsyncTCPClient:
    """异步TCP客户端"""

    def __init__(self, host='localhost', port=9000):
        self.host = host
        self.port = port
        self.reader = None
        self.writer = None

    async def connect(self):
        """连接到服务器"""
        try:
            self.reader, self.writer = await asyncio.open_connection(self.host, self.port)
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 已连接到服务器 {self.host}:{self.port}")
            return True
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 连接服务器失败: {e}")
            return False

    async def send_message(self, message):
        """发送消息"""
        if not self.writer:
            return None

        try:
            self.writer.write(message.encode('utf-8'))
            await self.writer.drain()

            # 读取响应
            response = await self.reader.read(1024)
            return response.decode('utf-8').strip()
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送消息失败: {e}")
            return None

    async def listen_for_messages(self):
        """监听服务器消息"""
        try:
            while True:
                data = await self.reader.read(1024)
                if not data:
                    break
                message = data.decode('utf-8').strip()
                if message:
                    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服务器消息: {message}")
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 监听消息时发生错误: {e}")

    async def disconnect(self):
        """断开连接"""
        if self.writer:
            self.writer.close()
            await self.writer.wait_closed()
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 已断开连接")

async def run_server():
    """运行服务器"""
    server = AsyncTCPServer()
    try:
        await server.start_server()
    except KeyboardInterrupt:
        print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服务器正在关闭...")

async def run_single_client():
    """运行单个客户端"""
    client = AsyncTCPClient()

    if not await client.connect():
        return

    try:
        # 启动消息监听任务
        listen_task = asyncio.create_task(client.listen_for_messages())

        # 发送测试消息
        messages = [
            "Hello, Async Server!",
            "time",
            "clients",
            "broadcast:大家好，我是异步客户端！",
            "Python异步网络编程测试",
            "quit"
        ]

        for message in messages:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 发送消息: {message}")
            response = await client.send_message(message)
            if response:
                print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服务器响应: {response}")
            await asyncio.sleep(1)

        # 取消监听任务
        listen_task.cancel()
        try:
            await listen_task
        except asyncio.CancelledError:
            pass

    finally:
        await client.disconnect()

async def run_multiple_clients(num_clients=3):
    """运行多个并发客户端"""
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 启动 {num_clients} 个并发客户端")

    async def client_task(client_id):
        client = AsyncTCPClient()
        if not await client.connect():
            return

        try:
            # 发送一些测试消息
            for i in range(3):
                message = f"客户端 {client_id} 的消息 {i+1}"
                response = await client.send_message(message)
                if response:
                    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客户端 {client_id} 收到响应: {response}")
                await asyncio.sleep(0.5)

            # 发送广播消息
            broadcast_msg = f"broadcast:来自客户端 {client_id} 的广播"
            await client.send_message(broadcast_msg)
            await asyncio.sleep(1)

            # 断开连接
            await client.send_message("quit")
        finally:
            await client.disconnect()

    # 并发运行多个客户端
    tasks = [client_task(i+1) for i in range(num_clients)]
    await asyncio.gather(*tasks)

if __name__ == '__main__':
    import sys

    if len(sys.argv) > 1:
        if sys.argv[1] == 'server':
            asyncio.run(run_server())
        elif sys.argv[1] == 'client':
            asyncio.run(run_single_client())
        elif sys.argv[1] == 'multi':
            asyncio.run(run_multiple_clients())
        else:
            print("用法: python async_server_demo.py [server|client|multi]")
    else:
        print("用法: python async_server_demo.py [server|client|multi]")
        print("  server - 启动异步服务器")
        print("  client - 启动单个客户端")
        print("  multi  - 启动多个并发客户端")

运行异步网络编程示例¶

启动异步服务器：

python async_server_demo.py server

在另一个终端启动多个并发客户端：

python async_server_demo.py multi

异步HTTP客户端¶

使用aiohttp库进行异步HTTP请求：

import aiohttp
import asyncio

async def fetch_url(session, url):
    """异步获取URL内容"""
    try:
        async with session.get(url) as response:
            return await response.text()
    except Exception as e:
        return f"错误: {e}"

async def main():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2',
        'https://httpbin.org/delay/3'
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

        for i, result in enumerate(results):
            print(f"URL {i+1} 结果长度: {len(result)}")

# 运行异步HTTP客户端
asyncio.run(main())

11.7 网络安全¶

网络安全是网络编程中的重要主题，需要考虑各种安全威胁和防护措施。

常见网络安全威胁¶

1. 注入攻击¶

SQL注入：恶意SQL代码注入
命令注入：系统命令注入
代码注入：恶意代码注入

2. 跨站攻击¶

XSS（跨站脚本）：注入恶意脚本
CSRF（跨站请求伪造）：伪造用户请求

3. 拒绝服务攻击¶

DoS攻击：使服务不可用
DDoS攻击：分布式拒绝服务

安全编程实践¶

1. 输入验证¶

import re

def validate_email(email):
    """验证邮箱格式"""
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return re.match(pattern, email) is not None

def sanitize_input(user_input):
    """清理用户输入"""
    # 移除危险字符
    dangerous_chars = ['<', '>', '"', "'", '&', ';']
    for char in dangerous_chars:
        user_input = user_input.replace(char, '')
    return user_input.strip()

2. 密码安全¶

import hashlib
import secrets

def generate_salt():
    """生成随机盐值"""
    return secrets.token_hex(16)

def hash_password(password, salt):
    """哈希密码"""
    return hashlib.pbkdf2_hmac('sha256', 
                              password.encode('utf-8'), 
                              salt.encode('utf-8'), 
                              100000)

def verify_password(password, salt, hashed):
    """验证密码"""
    return hash_password(password, salt) == hashed

SSL/TLS编程¶

Python的ssl模块提供了SSL/TLS支持：

import ssl
import socket

# 创建SSL上下文
context = ssl.create_default_context()

# 创建安全连接
with socket.create_connection(('www.python.org', 443)) as sock:
    with context.wrap_socket(sock, server_hostname='www.python.org') as ssock:
        print(f"SSL版本: {ssock.version()}")
        print(f"加密套件: {ssock.cipher()}")

        # 发送HTTPS请求
        ssock.send(b"GET / HTTP/1.1\r\nHost: www.python.org\r\n\r\n")
        response = ssock.recv(4096)
        print(response.decode('utf-8')[:200])

认证和授权¶

JWT令牌示例¶

import jwt
import datetime

# 生成JWT令牌
def generate_token(user_id, secret_key):
    payload = {
        'user_id': user_id,
        'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=24),
        'iat': datetime.datetime.utcnow()
    }
    return jwt.encode(payload, secret_key, algorithm='HS256')

# 验证JWT令牌
def verify_token(token, secret_key):
    try:
        payload = jwt.decode(token, secret_key, algorithms=['HS256'])
        return payload['user_id']
    except jwt.ExpiredSignatureError:
        return None  # 令牌已过期
    except jwt.InvalidTokenError:
        return None  # 无效令牌

总结¶

本章详细介绍了Python网络编程的各个方面：

网络编程基础：了解了TCP/IP协议栈、客户端-服务器模型和IP地址端口的概念
Socket编程：掌握了TCP和UDP Socket编程的实现方法，包括服务器端和客户端的开发
HTTP客户端编程：学习了使用urllib和requests库进行HTTP请求的方法
Web服务器开发：了解了如何使用http.server模块创建简单的Web服务器
网络数据处理：掌握了数据序列化、压缩和加密的基本技术
异步网络编程：学习了使用asyncio进行异步网络编程，提高应用性能
网络安全：了解了常见的网络安全威胁和防护措施

通过本章的学习，你应该能够：
- 理解网络编程的基本概念和原理
- 使用Socket进行底层网络通信
- 开发HTTP客户端和服务器应用
- 处理网络数据的序列化和安全问题
- 使用异步编程提高网络应用性能
- 实施基本的网络安全措施

网络编程是一个广阔的领域，本章提供了坚实的基础。在实际项目中，你可能还需要学习特定的Web框架（如Django、Flask）、消息队列、微服务架构等更高级的主题。记住，网络编程的关键是理解协议、处理并发和确保安全性。

第11章 网络编程¶

11.1 网络编程基础¶

网络协议概述¶

TCP/IP协议栈¶

常用网络协议¶

客户端-服务器模型¶

IP地址和端口¶

IP地址¶

端口号¶

11.2 Socket编程¶

Socket概念¶

套接字类型¶

Python的socket模块¶

地址族¶

TCP Socket编程实战¶

TCP服务器端¶

TCP客户端¶

运行TCP示例¶

UDP Socket编程实战¶

UDP服务器端¶

UDP客户端¶

运行UDP示例¶

TCP vs UDP 对比¶

Socket编程最佳实践¶

1. 异常处理¶

2. 超时设置¶

3. 地址重用¶

4. 缓冲区管理¶

11.3 HTTP客户端编程¶

HTTP协议基础¶

HTTP请求方法¶

HTTP状态码¶

urllib模块¶

urllib.request基本使用¶

requests库¶

安装requests¶

requests基本使用¶

运行HTTP客户端示例¶

requests高级功能¶

会话管理¶

文件上传¶

超时和重试¶

11.4 Web服务器开发¶

HTTP服务器基础¶

http.server模块¶

简单HTTP服务器¶

运行HTTP服务器示例¶

WSGI协议¶

WSGI应用示例¶

11.5 网络数据处理¶

数据序列化¶

JSON序列化¶

pickle序列化¶

数据压缩¶

gzip压缩¶

数据加密¶

hashlib哈希¶

11.6 异步网络编程¶

异步编程概念¶

同步vs异步¶

事件循环¶

asyncio模块¶

基本概念¶

异步网络编程实战¶

运行异步网络编程示例¶

异步HTTP客户端¶

11.7 网络安全¶

常见网络安全威胁¶

1. 注入攻击¶

2. 跨站攻击¶

3. 拒绝服务攻击¶

安全编程实践¶

1. 输入验证¶

2. 密码安全¶

SSL/TLS编程¶

认证和授权¶

JWT令牌示例¶

总结¶

相关文章

第11章网络编程¶