基于Python+socket的聊天管理系统

import socket import threading class ChatServer: def __init__(self, host="127.0.0.1", port=12345): self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server.bind((host, port)) self.server.listen(5) # 最大等待连接数 print(f"服务器启动，监听地址：{host}:{port}") self.clients = {} # 存储连接的客户端 {address: socket} def broadcast(self, message, exclude_socket=None): """向所有客户端广播消息""" for client_socket in self.clients.values(): if client_socket != exclude_socket: try: client_socket.send(message.encode("utf-8")) except: pass def handle_client(self, client_socket, address): """处理单个客户端连接""" print(f"客户端连接：{address}") self.clients[address] = client_socket while True: try: message = client_socket.recv(1024).decode("utf-8") if not message: break print(f"[{address}] 消息：{message}") # 广播消息给其他客户端 self.broadcast(f"[{address}] {message}", exclude_socket=client_socket) except: break # 客户端断开连接 print(f"客户端断开连接：{address}") client_socket.close() del self.clients[address] def run(self): """启动服务器""" while True: client_socket, address = self.server.accept() threading.Thread(target=self.handle_client, args=(client_socket, address)).start() if __name__ == "__main__": server = ChatServer() server.run()