### Python 实现三人聊天室（支持群聊和私聊） 以下是通过 `Python` 和 `Socket` 编程实现的一个简单三人聊天室的示例代码。该聊天室支持基本的群聊和私聊功能。 #### 功能描述 - 用户可以连接到服务器并加入房间。 - 支持用户之间的私聊消息传递。 - 所有用户的公共消息会广播给所有人。 --- #### 服务端代码 ```python import socket from threading import Thread clients = {} # 存储客户端套接字及其名称 addresses = {} # 存储客户端地址 HOST = '127.0.0.1' PORT = 9999 ADDR = (HOST, PORT) server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.bind(ADDR) def accept_connections(): """处理客户端连接""" while True: client, client_address = server_socket.accept() addresses[client] = client_address print(f"{client_address} 已连接.") client.send(bytes("请输入您的昵称:", "utf8")) Thread(target=handle_client, args=(client,)).start() def handle_client(client): """处理单个客户端的消息接收与分发""" name = client.recv(1024).decode("utf8") welcome_message = f"欢迎 {name} 加入聊天室!" broadcast(welcome_message.encode("utf8")) # 广播欢迎消息 clients[client] = name client.send(bytes("您可以开始聊天了.", "utf8")) while True: try: message = client.recv(1024) if not message.decode().startswith("@"): # 如果不是私聊消息，则广播 broadcast(message, name + ": ") else: # 处理私聊消息 private_chat(client, message.decode()) except Exception as e: del clients[client] broadcast(f"{name} 已离开聊天室.".encode("utf8")) break def broadcast(msg, prefix=""): """向所有客户端广播消息""" for sock in clients: sock.send(bytes(prefix, "utf8") + msg) def private_chat(sender_sock, raw_msg): """处理私聊逻辑""" parts = raw_msg.split(maxsplit=1) target_name = parts[0][1:] # 获取目标用户名 content = parts[1] found_target = False for sock, name in clients.items(): if name == target_name and sender_sock != sock: # 查找目标用户 sock.send(f"[私聊]{clients[sender_sock]} -> {target_name}: {content}".encode("utf8")) sender_sock.send(f"[私聊]-> {target_name}: {content}".encode("utf8")) found_target = True if not found_target: sender_sock.send(b"无法找到指定用户.") if __name__ == "__main__": server_socket.listen(5) print("等待连接...") ACCEPT_THREAD = Thread(target=accept_connections) ACCEPT_THREAD.start() ACCEPT_THREAD.join() server_socket.close() ``` --- #### 客户端代码 ```python import socket from threading import Thread HOST = '127.0.0.1' # 服务器IP地址 PORT = 9999 # 服务器端口 ADDR = (HOST, PORT) client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.connect(ADDR) def receive_messages(): """接收来自服务器的消息""" while True: try: message = client_socket.recv(1024).decode("utf8") print(message) except OSError: break def send_messages(): """发送消息至服务器""" while True: message = input("") client_socket.send(bytes(message, "utf8")) print("正在连接到服务器...") Thread(target=receive_messages).start() nickname = input("输入您的昵称: ") client_socket.send(nickname.encode("utf8")) send_messages_thread = Thread(target=send_messages) send_messages_thread.start() ``` --- #### 运行说明 1. **启动服务端**：先运行服务端代码，监听客户端连接。 2. **启动多个客户端**：分别打开三个终端窗口，依次运行客户端代码，并输入不同的昵称。 3. **测试功能**： - 输入普通消息即可进行群聊。 - 使用 `@<目标昵称> <消息>` 的格式可发起私聊。 --- ### 技术细节 - **Socket通信**：使用 TCP 协议建立可靠的双向通信通道[^1]。 - **线程管理**：每个客户端都由独立的线程负责消息收发，确保并发能力[^2]。 - **私聊机制**：通过解析消息前缀判断是否为目标用户发送私聊消息[^3]。 ---