Python教程

Appearance

Python 网络编程 ​

网络编程是 Python 中的一个重要领域，它允许程序通过网络与其他设备或服务进行通信。本章节将详细介绍 Python 中的网络编程概念、库和使用方法。

网络编程基础 ​

什么是网络编程？ ​

网络编程是指编写程序，使不同设备或进程之间通过网络进行数据交换和通信的过程。它涉及到网络协议、套接字编程、客户端-服务器模型等概念。

基本概念 ​

  1. IP 地址：标识网络中的设备
  2. 端口：标识设备上的服务
  3. 套接字（Socket）：网络通信的端点
  4. 协议：通信规则的集合，如 TCP、UDP、HTTP 等
  5. 客户端-服务器模型：一种常见的网络通信模式

网络协议 ​

  1. TCP（传输控制协议）：面向连接的、可靠的、基于字节流的传输层协议
  2. UDP（用户数据报协议）：无连接的、不可靠的、基于数据报的传输层协议
  3. HTTP（超文本传输协议）：应用层协议，用于 Web 通信
  4. HTTPS：HTTP 的安全版本，使用 TLS/SSL 加密
  5. FTP（文件传输协议）：用于文件传输
  6. SMTP（简单邮件传输协议）：用于发送电子邮件

套接字编程 ​

套接字是网络编程的基础，Python 提供了 socket 模块来支持套接字编程。

TCP 套接字 ​

TCP 服务器 ​

python
import socket
import threading

# 处理客户端连接
def handle_client(client_socket, client_address):
    """处理客户端连接"""
    print(f"接受到来自 {client_address} 的连接")
    
    try:
        # 接收数据
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            
            # 打印接收到的数据
            print(f"从 {client_address} 收到: {data.decode('utf-8')}")
            
            # 发送响应
            response = f"服务器收到: {data.decode('utf-8')}"
            client_socket.sendall(response.encode('utf-8'))
    finally:
        # 关闭连接
        client_socket.close()
        print(f"与 {client_address} 的连接已关闭")

# 创建 TCP 服务器
def start_tcp_server():
    """启动 TCP 服务器"""
    # 创建 TCP 套接字
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # 设置端口复用
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # 绑定地址和端口
    server_socket.bind(('localhost', 8080))
    
    # 开始监听
    server_socket.listen(5)
    print("TCP 服务器启动，监听端口 8080")
    
    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()
    finally:
        # 关闭服务器
        server_socket.close()

# 启动服务器
if __name__ == "__main__":
    start_tcp_server()

TCP 客户端 ​

python
import socket

# 创建 TCP 客户端
def start_tcp_client():
    """启动 TCP 客户端"""
    # 创建 TCP 套接字
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        # 连接服务器
        client_socket.connect(('localhost', 8080))
        print("连接到服务器成功")
        
        # 发送数据
        while True:
            message = input("请输入消息（输入 'exit' 退出）: ")
            if message == 'exit':
                break
            
            client_socket.sendall(message.encode('utf-8'))
            
            # 接收响应
            data = client_socket.recv(1024)
            print(f"从服务器收到: {data.decode('utf-8')}")
    except ConnectionRefusedError:
        print("连接服务器失败，请确保服务器已启动")
    finally:
        # 关闭连接
        client_socket.close()
        print("客户端连接已关闭")

# 启动客户端
if __name__ == "__main__":
    start_tcp_client()

UDP 套接字 ​

UDP 服务器 ​

python
import socket

# 创建 UDP 服务器
def start_udp_server():
    """启动 UDP 服务器"""
    # 创建 UDP 套接字
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
    # 设置端口复用
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # 绑定地址和端口
    server_socket.bind(('localhost', 8080))
    print("UDP 服务器启动，监听端口 8080")
    
    try:
        while True:
            # 接收数据
            data, client_address = server_socket.recvfrom(1024)
            print(f"从 {client_address} 收到: {data.decode('utf-8')}")
            
            # 发送响应
            response = f"服务器收到: {data.decode('utf-8')}"
            server_socket.sendto(response.encode('utf-8'), client_address)
    finally:
        # 关闭服务器
        server_socket.close()

# 启动服务器
if __name__ == "__main__":
    start_udp_server()

UDP 客户端 ​

python
import socket

# 创建 UDP 客户端
def start_udp_client():
    """启动 UDP 客户端"""
    # 创建 UDP 套接字
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
    try:
        server_address = ('localhost', 8080)
        print(f"准备向 {server_address} 发送数据")
        
        # 发送数据
        while True:
            message = input("请输入消息（输入 'exit' 退出）: ")
            if message == 'exit':
                break
            
            client_socket.sendto(message.encode('utf-8'), server_address)
            
            # 接收响应
            data, server = client_socket.recvfrom(1024)
            print(f"从服务器收到: {data.decode('utf-8')}")
    finally:
        # 关闭连接
        client_socket.close()
        print("客户端连接已关闭")

# 启动客户端
if __name__ == "__main__":
    start_udp_client()

HTTP 客户端 ​

Python 提供了多个库来进行 HTTP 客户端操作，包括 urllib、requests 等。

使用 urllib ​

python
import urllib.request
import urllib.parse
import urllib.error

# 发送 GET 请求
def send_get_request():
    """发送 GET 请求"""
    try:
        # 发送请求
        response = urllib.request.urlopen('https://www.example.com')
        
        # 读取响应
        data = response.read()
        print(f"响应状态码: {response.status}")
        print(f"响应头: {response.getheaders()}")
        print(f"响应内容: {data.decode('utf-8')}")
    except urllib.error.URLError as e:
        print(f"URL 错误: {e}")
    except urllib.error.HTTPError as e:
        print(f"HTTP 错误: {e.code} - {e.reason}")

# 发送 POST 请求
def send_post_request():
    """发送 POST 请求"""
    try:
        # 准备数据
        data = urllib.parse.urlencode({'name': 'Alice', 'age': 30}).encode('utf-8')
        
        # 创建请求
        req = urllib.request.Request('https://httpbin.org/post', data=data, method='POST')
        req.add_header('Content-Type', 'application/x-www-form-urlencoded')
        
        # 发送请求
        response = urllib.request.urlopen(req)
        
        # 读取响应
        data = response.read()
        print(f"响应状态码: {response.status}")
        print(f"响应内容: {data.decode('utf-8')}")
    except urllib.error.URLError as e:
        print(f"URL 错误: {e}")
    except urllib.error.HTTPError as e:
        print(f"HTTP 错误: {e.code} - {e.reason}")

# 测试 HTTP 请求
if __name__ == "__main__":
    print("发送 GET 请求:")
    send_get_request()
    print("\n发送 POST 请求:")
    send_post_request()

使用 requests ​

requests 是一个第三方库，提供了更简洁、更强大的 HTTP 客户端功能。

安装 requests ​

bash
pip install requests

使用示例 ​

python
import requests

# 发送 GET 请求
def send_get_request():
    """发送 GET 请求"""
    try:
        # 发送请求
        response = requests.get('https://www.example.com')
        
        # 处理响应
        print(f"响应状态码: {response.status_code}")
        print(f"响应头: {dict(response.headers)}")
        print(f"响应内容: {response.text}")
    except requests.RequestException as e:
        print(f"请求错误: {e}")

# 发送 POST 请求
def send_post_request():
    """发送 POST 请求"""
    try:
        # 准备数据
        data = {'name': 'Alice', 'age': 30}
        
        # 发送请求
        response = requests.post('https://httpbin.org/post', data=data)
        
        # 处理响应
        print(f"响应状态码: {response.status_code}")
        print(f"响应内容: {response.json()}")  # 自动解析 JSON
    except requests.RequestException as e:
        print(f"请求错误: {e}")

# 发送带参数的 GET 请求
def send_get_with_params():
    """发送带参数的 GET 请求"""
    try:
        # 准备参数
        params = {'q': 'python', 'page': 1}
        
        # 发送请求
        response = requests.get('https://httpbin.org/get', params=params)
        
        # 处理响应
        print(f"响应状态码: {response.status_code}")
        print(f"响应内容: {response.json()}")
    except requests.RequestException as e:
        print(f"请求错误: {e}")

# 发送带 headers 的请求
def send_with_headers():
    """发送带 headers 的请求"""
    try:
        # 准备 headers
        headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
        
        # 发送请求
        response = requests.get('https://httpbin.org/headers', headers=headers)
        
        # 处理响应
        print(f"响应状态码: {response.status_code}")
        print(f"响应内容: {response.json()}")
    except requests.RequestException as e:
        print(f"请求错误: {e}")

# 测试 HTTP 请求
if __name__ == "__main__":
    print("发送 GET 请求:")
    send_get_request()
    print("\n发送 POST 请求:")
    send_post_request()
    print("\n发送带参数的 GET 请求:")
    send_get_with_params()
    print("\n发送带 headers 的请求:")
    send_with_headers()

HTTP 服务器 ​

Python 提供了多个库来创建 HTTP 服务器，包括 http.server、socketserver、Flask、Django 等。

使用 http.server ​

http.server 是 Python 标准库中的模块，可以快速创建一个简单的 HTTP 服务器。

python
import http.server
import socketserver
import threading
import time

# 启动 HTTP 服务器
def start_http_server():
    """启动 HTTP 服务器"""
    PORT = 8000
    
    # 创建请求处理器
    Handler = http.server.SimpleHTTPRequestHandler
    
    # 创建服务器
    with socketserver.TCPServer(("", PORT), Handler) as httpd:
        print(f"HTTP 服务器启动，监听端口 {PORT}")
        print(f"访问地址: http://localhost:{PORT}")
        # 启动服务器
        httpd.serve_forever()

# 测试 HTTP 服务器
if __name__ == "__main__":
    # 启动服务器线程
    server_thread = threading.Thread(target=start_http_server)
    server_thread.daemon = True
    server_thread.start()
    
    # 等待用户输入
    input("按 Enter 键退出...\n")

使用 Flask ​

Flask 是一个轻量级的 Web 框架，用于创建更复杂的 HTTP 服务器。

安装 Flask ​

bash
pip install Flask

使用示例 ​

python
from flask import Flask, request, jsonify

# 创建 Flask 应用
app = Flask(__name__)

# 定义路由
@app.route('/')
def index():
    """首页"""
    return "Hello, World!"

# 定义带参数的路由
@app.route('/hello/<name>')
def hello(name):
    """带参数的路由"""
    return f"Hello, {name}!"

# 定义 POST 路由
@app.route('/api/data', methods=['POST'])
def api_data():
    """API 路由"""
    # 获取 JSON 数据
    data = request.get_json()
    if not data:
        return jsonify({"error": "No JSON data provided"}), 400
    
    # 处理数据
    name = data.get('name', 'Guest')
    age = data.get('age', 0)
    
    # 返回响应
    return jsonify({
        "message": "Data received",
        "name": name,
        "age": age
    })

# 启动服务器
if __name__ == "__main__":
    print("Flask 服务器启动，监听端口 5000")
    print("访问地址: http://localhost:5000")
    app.run(debug=True)

异步网络编程 ​

Python 3.4+ 引入了 asyncio 模块，支持异步网络编程，提高 I/O 密集型任务的性能。

使用 asyncio 和 aiohttp ​

aiohttp 是一个基于 asyncio 的异步 HTTP 客户端/服务器库。

安装 aiohttp ​

bash
pip install aiohttp

异步 HTTP 客户端 ​

python
import asyncio
import aiohttp

async def fetch(session, url):
    """异步获取 URL 内容"""
    async with session.get(url) as response:
        return await response.text()

async def main():
    """主协程函数"""
    urls = [
        "https://www.example.com",
        "https://www.python.org",
        "https://www.github.com"
    ]
    
    async with aiohttp.ClientSession() as session:
        # 并发执行多个请求
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        for url, content in zip(urls, results):
            print(f"URL: {url}, 内容长度: {len(content)}")

# 运行协程
if __name__ == "__main__":
    asyncio.run(main())

异步 HTTP 服务器 ​

python
from aiohttp import web
import asyncio

async def handle(request):
    """处理 HTTP 请求"""
    name = request.match_info.get('name', "Anonymous")
    await asyncio.sleep(0.5)  # 模拟 I/O 操作
    return web.Response(text=f"Hello, {name}!")

async def main():
    """主协程函数"""
    app = web.Application()
    app.add_routes([
        web.get('/', handle),
        web.get('/{name}', handle)
    ])
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    print("异步 HTTP 服务器启动，监听端口 8080")
    
    # 保持服务器运行
    while True:
        await asyncio.sleep(3600)

# 运行协程
if __name__ == "__main__":
    asyncio.run(main())

网络工具和库 ​

1. socket 模块 ​

  • 功能：提供底层套接字操作
  • 用途：创建 TCP/UDP 服务器和客户端
  • 示例：见前文的套接字编程部分

2. urllib 模块 ​

  • 功能：提供 HTTP 客户端功能
  • 组成：
    • urllib.request：发送 HTTP 请求
    • urllib.parse：解析 URL
    • urllib.error：处理异常
    • urllib.robotparser：解析 robots.txt
  • 示例：见前文的 HTTP 客户端部分

3. requests 库 ​

  • 功能：更简洁、更强大的 HTTP 客户端
  • 特点：自动处理会话、cookies、重定向等
  • 示例：见前文的 HTTP 客户端部分

4. http.server 模块 ​

  • 功能：创建简单的 HTTP 服务器
  • 用途：快速测试静态文件
  • 示例：见前文的 HTTP 服务器部分

5. Flask 框架 ​

  • 功能：轻量级 Web 框架
  • 用途：创建 RESTful API、Web 应用等
  • 示例：见前文的 HTTP 服务器部分

6. Django 框架 ​

  • 功能：全功能 Web 框架
  • 用途：创建复杂的 Web 应用
  • 特点：内置 ORM、admin 后台等

7. aiohttp 库 ​

  • 功能：异步 HTTP 客户端/服务器
  • 基于：asyncio
  • 用途：高并发 HTTP 应用
  • 示例：见前文的异步网络编程部分

8. socketserver 模块 ​

  • 功能：提供 TCP/UDP 服务器框架
  • 用途：简化服务器创建
  • 示例：
python
import socketserver

class MyTCPHandler(socketserver.BaseRequestHandler):
    """TCP 请求处理器"""
    
    def handle(self):
        # 接收数据
        self.data = self.request.recv(1024).strip()
        print(f"从 {self.client_address} 收到: {self.data.decode('utf-8')}")
        # 发送响应
        self.request.sendall(self.data.upper())

# 创建服务器
server = socketserver.TCPServer(('localhost', 8080), MyTCPHandler)
print("服务器启动，监听端口 8080")
# 启动服务器
server.serve_forever()

实际应用示例 ​

示例 1：简单的聊天服务器 ​

python
import socket
import threading

# 客户端列表
clients = []

# 广播消息
def broadcast(message, sender=None):
    """广播消息给所有客户端"""
    for client in clients:
        if client != sender:
            try:
                client.sendall(message)
            except:
                # 移除无效客户端
                clients.remove(client)

# 处理客户端连接
def handle_client(client_socket, client_address):
    """处理客户端连接"""
    print(f"接受到来自 {client_address} 的连接")
    clients.append(client_socket)
    
    try:
        # 发送欢迎消息
        welcome_message = f"欢迎来到聊天室，{client_address}！\n".encode('utf-8')
        client_socket.sendall(welcome_message)
        
        # 广播新用户加入
        join_message = f"{client_address} 加入了聊天室\n".encode('utf-8')
        broadcast(join_message, client_socket)
        
        # 接收消息
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            
            # 广播消息
            message = f"{client_address}: {data.decode('utf-8')}\n".encode('utf-8')
            broadcast(message, client_socket)
    finally:
        # 移除客户端
        if client_socket in clients:
            clients.remove(client_socket)
        
        # 广播用户离开
        leave_message = f"{client_address} 离开了聊天室\n".encode('utf-8')
        broadcast(leave_message)
        
        # 关闭连接
        client_socket.close()
        print(f"与 {client_address} 的连接已关闭")

# 创建聊天服务器
def start_chat_server():
    """启动聊天服务器"""
    # 创建 TCP 套接字
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # 设置端口复用
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # 绑定地址和端口
    server_socket.bind(('localhost', 8888))
    
    # 开始监听
    server_socket.listen(5)
    print("聊天服务器启动，监听端口 8888")
    print("客户端可以通过 telnet localhost 8888 连接")
    
    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()
    finally:
        # 关闭服务器
        server_socket.close()

# 启动服务器
if __name__ == "__main__":
    start_chat_server()

示例 2：简单的 HTTP 代理服务器 ​

python
import socket
import threading
import sys

def handle_client(client_socket):
    """处理客户端连接"""
    try:
        # 接收客户端请求
        request = client_socket.recv(4096)
        if not request:
            return
        
        # 解析请求行
        request_lines = request.split(b'\r\n')
        if not request_lines:
            return
        
        # 提取 URL
        request_line = request_lines[0]
        method, url, version = request_line.split(b' ', 2)
        
        # 解析 URL
        if url.startswith(b'http://'):
            url = url[7:]  # 移除 'http://'
        
        # 提取主机和端口
        host_port = url.split(b'/')[0]
        if b':' in host_port:
            host, port = host_port.split(b':')
            port = int(port)
        else:
            host = host_port
            port = 80
        
        # 创建与目标服务器的连接
        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_socket.connect((host.decode('utf-8'), port))
        
        # 发送请求到目标服务器
        server_socket.sendall(request)
        
        # 接收响应并转发给客户端
        while True:
            response = server_socket.recv(4096)
            if not response:
                break
            client_socket.sendall(response)
    except Exception as e:
        print(f"错误: {e}")
    finally:
        # 关闭连接
        if 'server_socket' in locals():
            server_socket.close()
        client_socket.close()

# 创建代理服务器
def start_proxy_server():
    """启动代理服务器"""
    # 创建 TCP 套接字
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # 设置端口复用
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # 绑定地址和端口
    server_socket.bind(('localhost', 8080))
    
    # 开始监听
    server_socket.listen(5)
    print("HTTP 代理服务器启动，监听端口 8080")
    print("设置浏览器代理为 localhost:8080")
    
    try:
        while True:
            # 接受连接
            client_socket, client_address = server_socket.accept()
            
            # 创建线程处理客户端
            client_thread = threading.Thread(
                target=handle_client, 
                args=(client_socket,)
            )
            client_thread.daemon = True
            client_thread.start()
    finally:
        # 关闭服务器
        server_socket.close()

# 启动服务器
if __name__ == "__main__":
    start_proxy_server()

示例 3：网络爬虫 ​

python
import requests
from bs4 import BeautifulSoup
import time

class SimpleCrawler:
    """简单的网络爬虫"""
    
    def __init__(self, start_url, max_depth=2):
        """初始化爬虫"""
        self.start_url = start_url
        self.max_depth = max_depth
        self.visited_urls = set()
    
    def crawl(self, url, depth=1):
        """爬取 URL"""
        if depth > self.max_depth:
            return
        
        if url in self.visited_urls:
            return
        
        print(f"爬取: {url} (深度: {depth})")
        self.visited_urls.add(url)
        
        try:
            # 发送请求
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            
            # 解析 HTML
            soup = BeautifulSoup(response.text, 'html.parser')
            
            # 提取链接
            links = []
            for a in soup.find_all('a', href=True):
                href = a['href']
                # 处理相对链接
                if href.startswith('http'):
                    links.append(href)
                elif href.startswith('/'):
                    base_url = '/'.join(url.split('/')[:3])
                    links.append(base_url + href)
            
            print(f"从 {url} 找到 {len(links)} 个链接")
            
            # 递归爬取子链接
            for link in links[:5]:  # 限制数量
                time.sleep(0.1)  # 避免请求过快
                self.crawl(link, depth + 1)
                
        except Exception as e:
            print(f"爬取 {url} 失败: {e}")
    
    def start(self):
        """开始爬取"""
        print(f"开始爬取，起始 URL: {self.start_url}")
        print(f"最大深度: {self.max_depth}")
        self.crawl(self.start_url)
        print(f"爬取完成，共访问 {len(self.visited_urls)} 个 URL")

# 测试爬虫
if __name__ == "__main__":
    crawler = SimpleCrawler("https://www.example.com", max_depth=2)
    crawler.start()

网络编程的最佳实践 ​

1. 错误处理 ​

  • 捕获异常：网络操作可能会失败，需要捕获和处理异常
  • 重试机制：对于临时性错误，实现重试机制
  • 超时设置：设置合理的超时时间，避免无限等待

2. 性能优化 ​

  • 使用连接池：重用数据库连接、HTTP 连接等
  • 异步编程：对于 I/O 密集型任务，使用异步编程
  • 缓存：缓存频繁访问的数据
  • 压缩：使用 gzip 等压缩算法减少数据传输量

3. 安全性 ​

  • 验证输入：验证所有用户输入，防止注入攻击
  • 使用 HTTPS：在生产环境中使用 HTTPS
  • 设置合理的权限：限制网络服务的权限
  • 防止 DDoS：实现速率限制等措施防止 DDoS 攻击

4. 可维护性 ​

  • 模块化：将网络代码模块化，提高可维护性
  • 日志记录：记录网络操作的日志，便于排查问题
  • 配置管理：使用配置文件管理网络参数
  • 文档：编写清晰的文档

5. 测试 ​

  • 单元测试：测试网络函数的功能
  • 集成测试：测试网络组件的集成
  • 性能测试：测试网络应用的性能
  • 安全性测试：测试网络应用的安全性

总结 ​

本章节介绍了 Python 中的网络编程，包括：

  1. 网络编程基础：IP 地址、端口、套接字、协议等基本概念
  2. 套接字编程：TCP 和 UDP 套接字的使用
  3. HTTP 客户端：使用 urllib 和 requests 发送 HTTP 请求
  4. HTTP 服务器：使用 http.server 和 Flask 创建 HTTP 服务器
  5. 异步网络编程：使用 asyncio 和 aiohttp 实现异步网络操作
  6. 网络工具和库：常用的网络编程库和工具
  7. 实际应用示例：聊天服务器、HTTP 代理服务器、网络爬虫
  8. 最佳实践：错误处理、性能优化、安全性、可维护性、测试

掌握 Python 中的网络编程，对于开发网络应用、API 客户端、Web 服务等非常重要。通过合理选择和使用网络编程库，可以提高开发效率和应用性能。

Copyright © 2025-2026 京公网安备 11010102000001号 京ICP备2024078128号-1