Python网络编程：从基础到高级应用¶

目录¶

网络编程基础
Socket编程
HTTP客户端编程
Web服务器开发
网络数据处理
异步网络编程
网络安全
总结

1. 网络编程基础¶

1.1 网络编程概述¶

网络编程是指编写程序使计算机之间能够通过网络进行通信和数据交换。Python提供了丰富的网络编程工具和库，使得开发人员能够轻松构建各种网络应用。

1.2 TCP/IP协议栈¶

TCP/IP协议栈是互联网通信的基础，分为四个主要层次：

应用层：HTTP、HTTPS、FTP、SMTP、DNS等协议
传输层：TCP（可靠传输）和UDP（不可靠但快速传输）
网络层：IP、ICMP、ARP等协议
链路层：以太网、WiFi等底层网络技术

1.3 客户端-服务器模型¶

客户端：发起请求的一方，如Web浏览器
服务器：响应请求的一方，如Web服务器
通信方式：基于请求-响应模式

1.4 IP地址和端口¶

IP地址：标识网络中的设备，IPv4（32位）或IPv6（128位）
端口：标识主机上的特定服务，0-65535范围，分为：
知名端口（0-1023）：HTTP(80)、HTTPS(443)等
注册端口（1024-49151）：应用程序注册使用
动态端口（49152-65535）：临时分配使用

2. Socket编程¶

2.1 Socket概念¶

Socket（套接字）是网络通信的端点，提供了进程间通信的接口。每个Socket由IP地址和端口号唯一标识。

2.2 TCP Socket编程¶

2.2.1 TCP服务器端¶

import socket
import threading

def handle_client(client_socket):
    """处理单个客户端连接"""
    client_address = client_socket.getpeername()
    print(f"新连接来自: {client_address}")

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

            message = data.decode('utf-8')
            print(f"收到: {message}")

            # 发送响应
            response = f"服务器已收到: {message}"
            client_socket.send(response.encode('utf-8'))

    except Exception as e:
        print(f"错误处理客户端: {e}")
    finally:
        client_socket.close()
        print(f"客户端 {client_address} 已断开连接")

def start_tcp_server(host='localhost', port=8888):
    """启动TCP服务器"""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
        # 设置地址重用，避免端口占用问题
        server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

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

        # 监听连接
        server_socket.listen(5)
        print(f"TCP服务器启动，监听 {host}:{port}")

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

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

        except KeyboardInterrupt:
            print("\n服务器正在关闭...")
        finally:
            server_socket.close()
            print("服务器已关闭")

if __name__ == "__main__":
    start_tcp_server()

2.2.2 TCP客户端¶

import socket

def start_tcp_client(host='localhost', port=8888):
    """启动TCP客户端"""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client_socket:
        try:
            # 连接服务器
            client_socket.connect((host, port))
            print(f"已连接到服务器 {host}:{port}")

            # 发送消息
            messages = [
                "Hello, Server!",
                "这是第二条消息",
                "再见！"
            ]

            for message in messages:
                client_socket.send(message.encode('utf-8'))
                print(f"发送: {message}")

                # 接收响应
                data = client_socket.recv(1024)
                if data:
                    print(f"收到: {data.decode('utf-8')}")

        except Exception as e:
            print(f"客户端错误: {e}")
        finally:
            print("客户端已关闭")

if __name__ == "__main__":
    start_tcp_client()

2.3 UDP Socket编程¶

UDP是无连接的传输协议，适用于实时性要求高但可靠性要求不高的场景。

2.3.1 UDP服务器端¶

import socket

def start_udp_server(host='localhost', port=9999):
    """启动UDP服务器"""
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server_socket:
        server_socket.bind((host, port))
        print(f"UDP服务器启动，监听 {host}:{port}")

        try:
            while True:
                # 接收数据和客户端地址
                data, client_address = server_socket.recvfrom(1024)
                message = data.decode('utf-8')
                print(f"收到来自 {client_address} 的消息: {message}")

                # 发送响应
                response = f"UDP服务器已收到: {message}"
                server_socket.sendto(response.encode('utf-8'), client_address)

        except KeyboardInterrupt:
            print("\n服务器正在关闭...")
        finally:
            server_socket.close()
            print("服务器已关闭")

if __name__ == "__main__":
    start_udp_server()

2.3.2 UDP客户端¶

import socket

def start_udp_client(host='localhost', port=9999):
    """启动UDP客户端"""
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client_socket:
        server_address = (host, port)

        try:
            # 发送消息
            messages = [
                "Hello, UDP Server!",
                "这是UDP第二条消息",
                "再见！"
            ]

            for message in messages:
                print(f"发送: {message}")
                client_socket.sendto(message.encode('utf-8'), server_address)

                # 接收响应
                data, server_address = client_socket.recvfrom(1024)
                print(f"收到: {data.decode('utf-8')}")

        except Exception as e:
            print(f"客户端错误: {e}")
        finally:
            print("客户端已关闭")

if __name__ == "__main__":
    start_udp_client()

2.4 TCP与UDP对比¶

特性	TCP	UDP
连接性	面向连接	无连接
可靠性	可靠（确认、重传）	不可靠（无重传机制）
速度	较慢	较快
有序性	保证数据按序到达	不保证
适用场景	文件传输、网页浏览	视频流、实时游戏

3. HTTP客户端编程¶

3.1 HTTP协议基础¶

HTTP（超文本传输协议）是应用层协议，用于传输超文本（网页、图片等）。

3.1.1 HTTP请求方法¶

GET：请求获取资源
POST：提交数据到服务器
PUT：更新资源
DELETE：删除资源

3.1.2 HTTP状态码¶

200 OK：请求成功
404 Not Found：资源不存在
500 Internal Server Error：服务器内部错误

3.2 使用urllib库¶

import urllib.request
import urllib.parse
import json

def fetch_data():
    """使用urllib获取数据"""
    url = 'https://httpbin.org/get'

    # 构建请求参数
    params = {
        'name': 'Python',
        'version': '3.9'
    }
    encoded_params = urllib.parse.urlencode(params).encode('utf-8')

    # 创建请求对象
    req = urllib.request.Request(
        url=url,
        data=encoded_params,
        method='GET'
    )

    try:
        # 发送请求
        with urllib.request.urlopen(req) as response:
            print(f"状态码: {response.getcode()}")
            print(f"响应头: {response.headers}")

            # 解析JSON响应
            data = json.loads(response.read().decode('utf-8'))
            print(json.dumps(data, indent=2))

    except urllib.error.URLError as e:
        print(f"请求错误: {e}")

if __name__ == "__main__":
    fetch_data()

3.3 使用requests库¶

requests库是更简洁易用的HTTP客户端库：

import requests
import json

def fetch_data_with_requests():
    """使用requests获取数据"""
    url = 'https://httpbin.org/get'

    # 请求参数
    params = {
        'name': 'Python',
        'version': '3.9'
    }

    try:
        # 发送GET请求
        response = requests.get(url, params=params, timeout=5)
        print(f"状态码: {response.status_code}")
        print(f"响应头: {response.headers}")

        # 解析JSON响应
        data = response.json()
        print(json.dumps(data, indent=2))

        # 发送POST请求
        post_url = 'https://httpbin.org/post'
        post_data = {
            'key1': 'value1',
            'key2': 'value2'
        }

        post_response = requests.post(post_url, json=post_data)
        print("\nPOST响应:")
        print(json.dumps(post_response.json(), indent=2))

    except requests.exceptions.RequestException as e:
        print(f"请求错误: {e}")

if __name__ == "__main__":
    fetch_data_with_requests()

3.4 高级请求处理¶

import requests

def advanced_requests():
    """高级请求处理示例"""
    url = 'https://httpbin.org/anything'

    # 自定义请求头
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
        'Accept': 'text/html,application/xhtml+xml,application/xml'
    }

    # 超时设置
    try:
        response = requests.get(
            url, 
            headers=headers, 
            timeout=5,
            verify=False  # 忽略SSL证书验证（仅测试用）
        )
        print(f"响应内容: {response.text[:200]}...")

        # 处理Cookie
        print(f"Cookie: {response.cookies.get_dict()}")

    except requests.exceptions.RequestException as e:
        print(f"请求错误: {e}")

if __name__ == "__main__":
    advanced_requests()

4. Web服务器开发¶

4.1 使用http.server模块¶

Python标准库提供了简单的HTTP服务器实现：

from http.server import BaseHTTPRequestHandler, HTTPServer
import time
import json

class SimpleHTTPRequestHandler(BaseHTTPRequestHandler):
    """自定义HTTP请求处理器"""

    def do_GET(self):
        """处理GET请求"""
        self.send_response(200)
        self.send_header('Content-Type', 'text/html; charset=utf-8')
        self.end_headers()

        # 根据路径返回不同内容
        if self.path == '/':
            response = """
            <html>
            <body>
                <h1>Python HTTP服务器</h1>
                <p>欢迎使用Python的简单HTTP服务器</p>
            </body>
            </html>
            """
        elif self.path == '/api/data':
            data = {
                'status': 'success',
                'message': '这是API数据',
                'timestamp': time.time()
            }
            response = json.dumps(data, ensure_ascii=False, indent=2)
            self.send_header('Content-Type', 'application/json')
        else:
            self.send_response(404)
            response = "页面未找到"

        self.wfile.write(response.encode('utf-8'))

    def log_message(self, format, *args):
        """禁用默认日志"""
        return

def run_server():
    """运行HTTP服务器"""
    server_address = ('localhost', 8000)
    httpd = HTTPServer(server_address, SimpleHTTPRequestHandler)
    print(f"服务器运行在 http://localhost:8000")

    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n服务器正在关闭...")
    finally:
        httpd.server_close()
        print("服务器已关闭")

if __name__ == "__main__":
    run_server()

4.2 使用WSGI协议¶

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

def application(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.get('PATH_INFO', '/')}</p>
        <p>方法: {environ.get('REQUEST_METHOD', 'GET')}</p>
        <p>查询参数: {environ.get('QUERY_STRING', '无')}</p>
    </body>
    </html>
    """

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

# 使用wsgiref运行服务器
from wsgiref.simple_server import make_server

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

5. 网络数据处理¶

5.1 数据序列化¶

5.1.1 JSON序列化¶

import json

data = {
    'name': '张三',
    'age': 25,
    'hobbies': ['阅读', '编程', '运动'],
    'is_student': True
}

# 转换为JSON字符串
json_str = json.dumps(data, ensure_ascii=False, indent=2)
print("JSON字符串:")
print(json_str)

# 转换回Python对象
parsed_data = json.loads(json_str)
print("\n解析后的数据:")
print(parsed_data)

5.1.2 使用pickle序列化¶

import pickle

data = {'name': '测试', 'numbers': [1, 2, 3, 4]}

# 序列化到文件
with open('data.pkl', 'wb') as f:
    pickle.dump(data, f)

# 从文件反序列化
with open('data.pkl', 'rb') as f:
    loaded_data = pickle.load(f)

print("加载的数据:", loaded_data)

5.2 数据压缩¶

import gzip
import json

data = {
    'message': '这是一个测试消息' * 1000  # 大量重复数据
}
json_data = json.dumps(data, ensure_ascii=False)

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

# 解压缩
decompressed_data = gzip.decompress(compressed_data).decode('utf-8')
original_data = json.loads(decompressed_data)
print("解压后数据是否正确:", original_data == data)

5.3 数据加密¶

5.3.1 使用hashlib进行哈希¶

```python
import hashlib

def hash_password(password):
“”“哈希密码”“”
salt = b’salt_value’ # 盐值
hashed = hashlib.pbkdf2_hmac(
‘sha256’,
password.encode(‘utf-8’),
salt,
100000 # 迭代次数
)
return hashed