Python 网络编程：socket 编程与 TCP/UDP 协议

网络编程是让程序通过网络与其他计算机通信的能力。Python 通过内置的 socket 模块，提供了对底层网络协议（如 TCP 和 UDP）的直接操作。掌握它，你就能编写聊天程序、文件传输工具或自定义网络服务。

理解 TCP 与 UDP 的核心区别

在动手写代码前，先明确两种协议的行为差异：

• TCP（Transmission Control Protocol）：像打电话。双方先建立连接，数据按顺序可靠送达，丢包会重传。适用于网页、邮件等不能出错的场景。
• UDP（User Datagram Protocol）：像发短信。直接发送数据包，不保证对方收到，也不保证顺序。但速度快、开销小，适合视频通话、游戏等容忍少量丢包的场景。

选择依据很简单：需要绝对可靠就用 TCP；追求速度且能容忍丢包就用 UDP。

准备工作：导入模块与基础概念

打开终端，确保已安装 Python（3.6+）。无需额外库，因为 socket 是标准库。

关键概念：

• IP 地址：标识网络中的设备，如 127.0.0.1（本机回环地址）。
• 端口（Port）：一个数字（0-65535），标识设备上的具体程序。例如 Web 服务常用 80，自定义程序建议用 8000 以上。
• 套接字（Socket）：网络通信的端点，由 IP + 端口唯一确定。

编写 TCP 服务器与客户端

步骤 1：创建 TCP 服务器

新建文件 tcp_server.py，写入以下代码：

import socket

# 创建 TCP 套接字（AF_INET 表示 IPv4，SOCK_STREAM 表示 TCP）
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 绑定到本地地址和端口
server_socket.bind(('127.0.0.1', 8888))

# 开始监听，最多允许 5 个连接排队
server_socket.listen(5)
print("TCP 服务器启动，监听 127.0.0.1:8888")

while True:
    # 接受客户端连接
    client_socket, client_address = server_socket.accept()
    print(f"新连接来自 {client_address}")

    # 接收数据（最大 1024 字节）
    data = client_socket.recv(1024)
    if data:
        message = data.decode('utf-8')
        print(f"收到消息: {message}")
        # 回复客户端
        client_socket.send(f"服务器已收到: {message}".encode('utf-8'))

    # 关闭客户端连接
    client_socket.close()

运行服务器：在终端执行 python tcp_server.py。

步骤 2：创建 TCP 客户端

新建文件 tcp_client.py，写入：

import socket

# 创建 TCP 套接字
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 连接到服务器
client_socket.connect(('127.0.0.1', 8888))

# 发送消息
message = "Hello from TCP client!"
client_socket.send(message.encode('utf-8'))

# 接收回复
response = client_socket.recv(1024)
print(response.decode('utf-8'))

# 关闭连接
client_socket.close()

测试通信：

1. 先运行 tcp_server.py（保持运行状态）。
2. 再打开新终端，运行 tcp_client.py。
3. 观察服务器终端输出 收到消息: Hello from TCP client!，客户端终端显示回复。

编写 UDP 服务器与客户端

UDP 无需建立连接，代码更简洁。

步骤 1：创建 UDP 服务器

新建文件 udp_server.py：

import socket

# 创建 UDP 套接字（SOCK_DGRAM 表示 UDP）
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 绑定地址和端口
server_socket.bind(('127.0.0.1', 9999))
print("UDP 服务器启动，监听 127.0.0.1:9999")

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

    # 直接回复（需指定客户端地址）
    reply = f"UDP 服务器已收到: {message}"
    server_socket.sendto(reply.encode('utf-8'), client_address)

运行服务器：python udp_server.py。

步骤 2：创建 UDP 客户端

新建文件 udp_client.py：

import socket

# 创建 UDP 套接字
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 发送消息到服务器（无需 connect）
server_address = ('127.0.0.1', 9999)
message = "Hello from UDP client!"
client_socket.sendto(message.encode('utf-8'), server_address)

# 接收回复
response, _ = client_socket.recvfrom(1024)
print(response.decode('utf-8'))

# 关闭套接字
client_socket.close()

测试通信：

1. 运行 udp_server.py。
2. 运行 udp_client.py。
3. 服务器显示收到消息，客户端显示回复。

关键参数与错误处理

实际项目中必须处理异常和资源释放。

设置超时避免卡死

添加超时：在套接字创建后调用 settimeout(seconds)。例如：

client_socket.settimeout(5.0)  # 5 秒无响应则抛出异常

使用 with 语句自动关闭

改写客户端（以 TCP 为例）：

import socket

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client_socket:
    client_socket.connect(('127.0.0.1', 8888))
    client_socket.send(b"Test message")
    response = client_socket.recv(1024)
    print(response.decode())
# 套接字在此自动关闭，即使发生异常

常见错误码处理

• ConnectionRefusedError：服务器未运行或端口错误。
• TimeoutError：网络延迟或对方无响应。
• OSError：端口被占用（常见于服务器重启太快）。

解决端口占用：在服务器绑定前添加：

server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

协议选择速查表

根据场景快速决策：

调试技巧：用 telnet 和 nc 测试

无需写客户端，用系统工具快速验证服务器。

测试 TCP 服务器：

1. 运行你的 tcp_server.py。
2. 打开新终端，执行 telnet 127.0.0.1 8888。
3. 输入任意文本并回车，观察服务器是否收到。

测试 UDP 服务器：

1. 运行 udp_server.py。
2. 执行 nc -u 127.0.0.1 9999（Linux/macOS）或 ncat -u 127.0.0.1 9999（Windows 需安装 Nmap）。
3. 输入文本，按 Ctrl+D（Linux）或 Ctrl+C（Windows）发送。

安全注意事项

• 永远不要在公网暴露未认证的服务：示例代码仅限本地测试。
• 验证输入数据：对 recv() 收到的数据做长度和内容检查，防止缓冲区溢出。
• 使用防火墙：生产环境中限制端口访问来源。

扩展：同时处理多个客户端（TCP）

基础服务器一次只能处理一个客户端。要支持并发，可用多线程：

import socket
import threading

def handle_client(client_socket, address):
    try:
        data = client_socket.recv(1024)
        if data:
            client_socket.send(b"Processed by thread")
    finally:
        client_socket.close()

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('127.0.0.1', 8888))
server.listen()

while True:
    client_sock, addr = server.accept()
    # 为每个客户端启动新线程
    client_thread = threading.Thread(
        target=handle_client,
        args=(client_sock, addr)
    )
    client_thread.start()

注意：高并发场景应使用异步 I/O（如 asyncio），但多线程适合入门理解。

验证你的代码

逐行检查：

1. TCP 服务器用 SOCK_STREAM，UDP 用 SOCK_DGRAM。
2. TCP 客户端必须先 connect()，UDP 直接 sendto()。
3. 所有字符串发送前用 .encode('utf-8')，接收后用 .decode('utf-8')。
4. 服务器绑定 127.0.0.1 仅限本机访问；若需局域网访问，改用 '0.0.0.0'。

运行顺序：先启动服务器，再运行客户端。若失败，检查端口号是否冲突（换用 10000+ 端口）。