前言
python实现TCP服务端，支持对所有客户端的数据收发，支持终端自定义命令操作，提供clear命令一键断开所有的客户端连接
简单易懂，直接上码
源码
import socket
import threadingclass TCPServer:# 修改此处ip 端口def __init__(self, host='0.0.0.0', port=12345):self.host = hostself.port = portself.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.clients = [] # 存储所有连接的客户端self.running = True # 标记服务器是否正在运行def start(self):self.server_socket.bind((self.host, self.port))self.server_socket.listen(5)print(f"服务器启动于 {self.host}:{self.port}")# 启动一个线程来监听控制台命令threading.Thread(target=self.console_listener, daemon=True).start()try:while self.running:client_socket, addr = self.server_socket.accept()print(f"接受来自 {addr} 的连接")self.clients.append(client_socket)threading.Thread(target=self.handle_client, args=(client_socket,), daemon=True).start()except KeyboardInterrupt:self.stop()def stop(self):print("停止服务器...")self.running = Falseself.server_socket.close()for client in self.clients:client.close()print("服务器已停止。")def handle_client(self, client_socket):try:while True:data = client_socket.recv(1024)if not data:breakprint(f"收到数据: {data.decode()}")client_socket.sendall(data) # 将接收到的数据回传给客户端except Exception as e:print(f"处理客户端时发生错误: {e}")finally:client_socket.close()if client_socket in self.clients:self.clients.remove(client_socket)print(f"客户端已断开连接。")def console_listener(self):while self.running:cmd = input("输入命令（输入 'clear' 断开所有客户端连接）: ")if cmd.strip().lower() == 'clear':print("断开所有客户端连接...")for client in list(self.clients):client.close()if client in self.clients:self.clients.remove(client)print("所有客户端已断开连接。")else:print("将消息发送给所有客户端...")for client in self.clients:try:client.sendall(cmd.encode())except Exception as e:print(f"发送消息时发生错误: {e}")if client in self.clients:self.clients.remove(client)print("消息已发送。")if __name__ == "__main__":server = TCPServer()server.start()
效果图
