Python网络编程与Socket通信一、Socket基础概念Socket套接字是网络通信的端点提供了进程间通信的机制。Python的socket模块提供了对底层网络接口的访问。import socket# 创建TCP Sockettcp_socket socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 创建UDP Socketudp_socket socket.socket(socket.AF_INET, socket.SOCK_DGRAM)# 地址族# AF_INET - IPv4# AF_INET6 - IPv6# AF_UNIX - Unix域套接字# Socket类型# SOCK_STREAM - TCP面向连接可靠# SOCK_DGRAM - UDP无连接不可靠但快速二、TCP服务器与客户端2.1 简单TCP服务器import socketimport threadingclass TCPServer:def __init__(self, host0.0.0.0, port8080):self.host hostself.port portself.server_socket socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)def start(self):self.server_socket.bind((self.host, self.port))self.server_socket.listen(5)print(f服务器启动在 {self.host}:{self.port})try:while True:client_socket, address self.server_socket.accept()print(f新连接来自 {address})thread threading.Thread(targetself.handle_client,args(client_socket, address))thread.daemon Truethread.start()except KeyboardInterrupt:print(服务器关闭)finally:self.server_socket.close()def handle_client(self, client_socket, address):try:while True:data client_socket.recv(4096)if not data:breakmessage data.decode(utf-8)print(f收到来自 {address}: {message})response f服务器收到: {message}client_socket.send(response.encode(utf-8))except ConnectionResetError:print(f客户端 {address} 断开连接)finally:client_socket.close()2.2 TCP客户端class TCPClient:def __init__(self, hostlocalhost, port8080):self.host hostself.port portself.socket socket.socket(socket.AF_INET, socket.SOCK_STREAM)def connect(self):self.socket.connect((self.host, self.port))print(f已连接到 {self.host}:{self.port})def send(self, message):self.socket.send(message.encode(utf-8))response self.socket.recv(4096)return response.decode(utf-8)def close(self):self.socket.close()# 使用client TCPClient()client.connect()response client.send(Hello, Server!)print(response)client.close()三、带协议的通信网络通信需要定义消息边界常见方案3.1 固定长度头部协议import structimport jsonclass MessageProtocol:消息格式4字节长度头 JSON消息体HEADER_SIZE 4staticmethoddef encode(data: dict) - bytes:body json.dumps(data).encode(utf-8)header struct.pack(!I, len(body)) # 大端序4字节无符号整数return header bodystaticmethoddef decode(socket_conn) - dict:# 读取头部header MessageProtocol._recv_exact(socket_conn, MessageProtocol.HEADER_SIZE)if not header:return Nonebody_length struct.unpack(!I, header)[0]# 读取消息体body MessageProtocol._recv_exact(socket_conn, body_length)return json.loads(body.decode(utf-8))staticmethoddef _recv_exact(conn, length):确保接收指定长度的数据data bwhile len(data) length:chunk conn.recv(length - len(data))if not chunk:return Nonedata chunkreturn data3.2 使用协议的服务器class ProtocolServer:def __init__(self, host0.0.0.0, port9000):self.host hostself.port portself.handlers {}def register_handler(self, action, handler):self.handlers[action] handlerdef start(self):with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)server.bind((self.host, self.port))server.listen(5)print(f协议服务器启动在 {self.host}:{self.port})while True:conn, addr server.accept()threading.Thread(targetself._handle, args(conn, addr)).start()def _handle(self, conn, addr):try:while True:message MessageProtocol.decode(conn)if message is None:breakaction message.get(action)handler self.handlers.get(action)if handler:response handler(message.get(data))else:response {error: f未知操作: {action}}conn.send(MessageProtocol.encode(response))finally:conn.close()# 注册处理器server ProtocolServer()server.register_handler(echo, lambda data: {echo: data})server.register_handler(time, lambda data: {time: str(datetime.now())})四、UDP通信import socketclass UDPServer:def __init__(self, host0.0.0.0, port8888):self.socket socket.socket(socket.AF_INET, socket.SOCK_DGRAM)self.socket.bind((host, port))print(fUDP服务器启动在 {host}:{port})def run(self):while True:data, addr self.socket.recvfrom(65535)message data.decode(utf-8)print(f收到来自 {addr}: {message})response fACK: {message}self.socket.sendto(response.encode(utf-8), addr)class UDPClient:def __init__(self):self.socket socket.socket(socket.AF_INET, socket.SOCK_DGRAM)self.socket.settimeout(5.0) # 设置超时def send(self, message, hostlocalhost, port8888):self.socket.sendto(message.encode(utf-8), (host, port))try:data, addr self.socket.recvfrom(65535)return data.decode(utf-8)except socket.timeout:return None五、select/poll多路复用import selectclass SelectServer:使用select实现非阻塞IO多路复用def __init__(self, host0.0.0.0, port8080):self.server socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server.setblocking(False)self.server.bind((host, port))self.server.listen(100)self.inputs [self.server]self.outputs []self.message_queues {}def run(self):print(Select服务器启动)while self.inputs:readable, writable, exceptional select.select(self.inputs, self.outputs, self.inputs, 1.0)for sock in readable:if sock is self.server:# 新连接conn, addr sock.accept()conn.setblocking(False)self.inputs.append(conn)self.message_queues[conn] []else:# 已有连接的数据data sock.recv(4096)if data:self.message_queues[sock].append(data)if sock not in self.outputs:self.outputs.append(sock)else:self._remove_connection(sock)for sock in writable:if self.message_queues.get(sock):msg self.message_queues[sock].pop(0)sock.send(msg)else:self.outputs.remove(sock)for sock in exceptional:self._remove_connection(sock)def _remove_connection(self, sock):if sock in self.inputs:self.inputs.remove(sock)if sock in self.outputs:self.outputs.remove(sock)del self.message_queues[sock]sock.close()六、selectors高级接口import selectorsclass SelectorServer:使用selectors模块推荐方式def __init__(self, host0.0.0.0, port8080):self.sel selectors.DefaultSelector()self.server socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server.setblocking(False)self.server.bind((host, port))self.server.listen(100)self.sel.register(self.server, selectors.EVENT_READ, self._accept)def _accept(self, sock, mask):conn, addr sock.accept()conn.setblocking(False)self.sel.register(conn, selectors.EVENT_READ, self._read)def _read(self, conn, mask):data conn.recv(4096)if data:conn.send(data) # echoelse:self.sel.unregister(conn)conn.close()def run(self):print(Selector服务器启动)while True:events self.sel.select(timeout1)for key, mask in events:callback key.datacallback(key.fileobj, mask)七、HTTP客户端实现class SimpleHTTPClient:手动实现HTTP/1.1客户端def __init__(self):self.socket Nonedef request(self, method, url, headersNone, bodyNone):from urllib.parse import urlparseparsed urlparse(url)host parsed.hostnameport parsed.port or 80path parsed.path or /# 建立连接self.socket socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.socket.connect((host, port))# 构建请求request_line f{method} {path} HTTP/1.1\r\ndefault_headers {Host: host,Connection: close,User-Agent: PythonSocket/1.0,}if headers:default_headers.update(headers)if body:default_headers[Content-Length] str(len(body))header_str .join(f{k}: {v}\r\n for k, v in default_headers.items())request f{request_line}{header_str}\r\nif body:request bodyself.socket.send(request.encode(utf-8))# 接收响应response bwhile True:chunk self.socket.recv(4096)if not chunk:breakresponse chunkself.socket.close()return self._parse_response(response)def _parse_response(self, raw):header_end raw.find(b\r\n\r\n)header_part raw[:header_end].decode(utf-8)body raw[header_end 4:]lines header_part.split(\r\n)status_line lines[0]status_code int(status_line.split( )[1])headers {}for line in lines[1:]:key, value line.split(: , 1)headers[key.lower()] valuereturn {status: status_code, headers: headers, body: body}八、WebSocket简易实现import hashlibimport base64class WebSocketServer:MAGIC_STRING 258EAFA5-E914-47DA-95CA-C5AB0DC85B11def __init__(self, host0.0.0.0, port8765):self.server socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server.bind((host, port))self.server.listen(5)def _handshake(self, conn):WebSocket握手data conn.recv(4096).decode(utf-8)key Nonefor line in data.split(\r\n):if Sec-WebSocket-Key in line:key line.split(: )[1].strip()breakif not key:return Falseaccept_key base64.b64encode(hashlib.sha1((key self.MAGIC_STRING).encode()).digest()).decode()response (HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nfSec-WebSocket-Accept: {accept_key}\r\n\r\n)conn.send(response.encode())return Truedef _decode_frame(self, data):解码WebSocket帧if len(data) 2:return Noneopcode data[0] 0x0Fmasked data[1] 0x80payload_length data[1] 0x7Foffset 2if payload_length 126:payload_length struct.unpack(!H, data[2:4])[0]offset 4elif payload_length 127:payload_length struct.unpack(!Q, data[2:10])[0]offset 10if masked:mask data[offset:offset4]offset 4payload bytearray(data[offset:offsetpayload_length])for i in range(len(payload)):payload[i] ^ mask[i % 4]else:payload data[offset:offsetpayload_length]return {opcode: opcode, payload: bytes(payload)}def _encode_frame(self, message):编码WebSocket帧payload message.encode(utf-8)frame bytearray()frame.append(0x81) # FIN text opcodelength len(payload)if length 126:frame.append(length)elif length 65536:frame.append(126)frame.extend(struct.pack(!H, length))else:frame.append(127)frame.extend(struct.pack(!Q, length))frame.extend(payload)return bytes(frame)九、实用工具函数def get_local_ip():获取本机IP地址s socket.socket(socket.AF_INET, socket.SOCK_DGRAM)try:s.connect((8.8.8.8, 80))return s.getsockname()[0]finally:s.close()def port_scan(host, start_port, end_port, timeout0.5):端口扫描open_ports []for port in range(start_port, end_port 1):sock socket.socket(socket.AF_INET, socket.SOCK_STREAM)sock.settimeout(timeout)result sock.connect_ex((host, port))if result 0:open_ports.append(port)sock.close()return open_portsdef resolve_dns(hostname):DNS解析try:ip socket.gethostbyname(hostname)all_ips socket.gethostbyname_ex(hostname)return {ip: ip, all: all_ips}except socket.gaierror as e:return {error: str(e)}十、总结网络编程要点1. TCP适合需要可靠传输的场景UDP适合实时性要求高的场景2. 必须处理消息边界问题TCP是字节流不保证消息完整性3. 生产环境使用selectors或asyncio处理并发连接4. 注意设置超时避免永久阻塞5. 实际项目中优先使用高层库requests、aiohttp、websockets