【Python socket模块深度解析】网络通信的核心工具

---

🌟 模块概述

🧬 模块定位

socket模块是Python标准库中用于网络通信的核心模块，提供对BSD套接字接口的面向对象封装，支持TCP/IP、UDP等协议栈。

📦 导入方式

import socket  # 标准导入
from socket import socket, AF_INET, SOCK_STREAM  # 按需导入

---

🛠️ 核心API详解

1️⃣ socket对象创建

socket.socket(family=AF_INET, type=SOCK_STREAM, proto=0, fileno=None)

• 参数说明：
  • family：地址族（默认AF_INET=IPv4，可选AF_INET6=IPv6）
  • type：套接字类型（SOCK_STREAM=TCP，SOCK_DGRAM=UDP）
  • proto：协议号（通常自动检测）
  • fileno：文件描述符（高级用法）

2️⃣ 关键方法速查

方法	适用协议	作用描述
bind((host, port))	TCP/UDP	绑定地址到套接字
listen(backlog)	TCP	启动监听，设置等待队列长度
accept()	TCP	接受客户端连接，返回(client_socket, address)
connect((host, port))	TCP	连接服务端
send(data)	TCP	发送数据
recv(bufsize)	TCP	接收数据
sendto(data, addr)	UDP	发送数据到指定地址
recvfrom(bufsize)	UDP	接收数据和发送方地址
close()	通用	关闭套接字

---

🔌 TCP/UDP编程流程对比

🚥 TCP通信流程

🚀 UDP通信流程

---

💻 实战代码示例

案例1：TCP文件传输服务器

import socketdef tcp_file_server(filename, port=8888):with socket.socket() as s:s.bind(('', port))s.listen(1)print(f"等待连接，端口{port}...")conn, addr = s.accept()with conn, open(filename, 'wb') as f:print(f"客户端 {addr} 已连接")while True:data = conn.recv(4096)if not data: breakf.write(data)print("文件传输完成")tcp_file_server('received_file.zip')

案例2：UDP时间客户端

import socket
import timedef udp_time_client(server_ip='127.0.0.1', port=8888):with socket.socket(type=socket.SOCK_DGRAM) as s:for _ in range(3):s.sendto(b'time', (server_ip, port))data, addr = s.recvfrom(1024)print(f"服务器 {addr} 时间: {data.decode()}")time.sleep(1)udp_time_client()

---

⚠️ 常见错误处理

1️⃣ 地址已在使用（Address already in use）

s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # 在bind前设置

2️⃣ 连接重置（Connection reset）

try:data = conn.recv(1024)
except ConnectionResetError:print("客户端异常断开")

3️⃣ 超时控制

s.settimeout(5.0)  # 设置5秒超时
try:data = s.recv(1024)
except socket.timeout:print("接收超时")

---

🚀 高级技巧

1️⃣ 获取本机IP

def get_local_ip():s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)try:s.connect(('8.8.8.8', 80))  # 连接公共DNSreturn s.getsockname()[0]finally:s.close()print("本机IP:", get_local_ip())

2️⃣ 端口扫描工具

def port_scan(target_ip, start=70, end=100):open_ports = []for port in range(start, end+1):with socket.socket() as s:s.settimeout(0.5)result = s.connect_ex((target_ip, port))if result == 0:open_ports.append(port)return open_portsprint("开放端口:", port_scan('127.0.0.1', 70, 100))

---

📚 学习资源推荐

1. 官方文档：socket — Low-level networking interface
2. 经典书籍：《Python网络编程（第3版）》
3. 调试工具：
   • Wireshark（网络协议分析）
   • netcat（网络调试瑞士军刀）
   • Postman（API测试）

“Socket编程是打开网络世界的钥匙，掌握它意味着你能让任何设备对话。”
—— 网络工程师箴言