基于Python实现多人聊天室的示例代码

import socket
import threading
from datetime import datetime
from collections import ordereddict
 
class chatserver:
    def __init__(self, host='0.0.0.0', port=50000):
        self.clients = ordereddict()
        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)
        print(f"🟢 服务端已启动在 {host}:{port}")
 
    def handle_client(self, conn, addr):
        username = none
        try:
            username = conn.recv(1024).decode().strip()
            if not username or username in self.clients:
                conn.send("username_invalid".encode())
                conn.close()
                return
 
            conn.send("connect_success".encode())
            self.clients[username] = conn
            self.broadcast_system_msg(f"🚩 {username} 进入聊天室")
            self.broadcast_userlist()
 
            while true:
                try:
                    data = conn.recv(4096)
                    if not data: break
                    
                    msg = data.decode().strip()
                    if msg == "heartbeat":
                        conn.send(b"heartbeat_ack")
                    elif msg.startswith("@@"):
                        target, _, content = msg[2:].partition(' ')
                        self.handle_private(username, target, content)
                    else:
                        self.broadcast_msg(f"{username}：{msg}")
 
                except exception as e:
                    print(f"处理错误：{str(e)}")
                    break
 
        except connectionreseterror:
            print(f"❌ {username} 异常断开")
        finally:
            if username in self.clients:
                del self.clients[username]
                self.broadcast_system_msg(f"🚩 {username} 离开聊天室")
                self.broadcast_userlist()
            conn.close()
 
    def handle_private(self, sender, target, content):
        if target in self.clients:
            timestamp = datetime.now().strftime("%h:%m:%s")
            msg = f"[{timestamp}] [私聊] {sender} -> 你：{content}"
            self.clients[target].send(msg.encode())
            self.clients[sender].send(msg.encode())
        else:
            self.clients[sender].send(f"用户 {target} 不在线".encode())
 
    def broadcast_msg(self, msg):
        timestamp = datetime.now().strftime("%h:%m:%s")
        full_msg = f"[{timestamp}] {msg}\n"
        for client in self.clients.values():
            try: client.send(full_msg.encode())
            except: pass
 
    def broadcast_system_msg(self, msg):
        full_msg = f"system:{msg}\n"
        for client in self.clients.values():
            try: client.send(full_msg.encode())
            except: pass
 
    def broadcast_userlist(self):
        user_list = ",".join(self.clients.keys())
        msg = f"userlist:{user_list}\n"
        for client in self.clients.values():
            try: client.send(msg.encode())
            except: pass
 
    def start(self):
        while true:
            conn, addr = self.server.accept()
            threading.thread(target=self.handle_client, args=(conn, addr)).start()
 
if __name__ == "__main__":
    chatserver().start() 

import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
import socket
import threading
 
class chatclient:
    def __init__(self, master):
        self.master = master
        self.client_socket = none
        self.username = ""
        self.running = false
        
        master.title("在线聊天室")
        master.geometry("900x600")
        self.create_widgets()
        self.show_login_dialog()
 
    def create_widgets(self):
        self.user_frame = ttk.frame(self.master, width=200)
        self.user_frame.pack(side=tk.left, fill=tk.y)
        
        self.user_list = ttk.treeview(self.user_frame, show="tree", selectmode='browse')
        self.user_list.pack(expand=true, fill=tk.both)
        self.user_list.bind('<<treeviewselect>>', self.select_user)
 
        self.chat_frame = ttk.frame(self.master)
        self.chat_frame.pack(expand=true, fill=tk.both)
        
        self.chat_area = scrolledtext.scrolledtext(self.chat_frame, state=tk.disabled)
        self.chat_area.pack(expand=true, fill=tk.both)
        
        # 输入框和按钮框架
        input_frame = ttk.frame(self.chat_frame)
        input_frame.pack(fill=tk.x, pady=5)
        
        self.msg_entry = ttk.entry(input_frame)
        self.msg_entry.pack(side=tk.left, expand=true, fill=tk.x)
        self.msg_entry.bind("<return>", self.send_message)
        
        # 添加发送按钮
        send_btn = ttk.button(input_frame, text="发送", command=self.send_message)
        send_btn.pack(side=tk.right, padx=5)
 
    def select_user(self, event):
        selected = self.user_list.selection()
        if selected:
            target = self.user_list.item(selected[0])['text']
            current = self.msg_entry.get()
            self.msg_entry.delete(0, tk.end)
            self.msg_entry.insert(0, f"@@{target} " if not current.startswith("@") else "")
 
    def show_login_dialog(self):
        self.login_dialog = tk.toplevel(self.master)
        self.login_dialog.title("登录")
        
        ttk.label(self.login_dialog, text="服务器地址:").grid(row=0, column=0, padx=5, pady=5)
        self.server_entry = ttk.entry(self.login_dialog)
        self.server_entry.insert(0, "127.0.0.1")
        self.server_entry.grid(row=0, column=1, padx=5, pady=5)
 
        ttk.label(self.login_dialog, text="端口号:").grid(row=1, column=0, padx=5, pady=5)
        self.port_entry = ttk.entry(self.login_dialog)
        self.port_entry.insert(0, "50000")
        self.port_entry.grid(row=1, column=1, padx=5, pady=5)
 
        ttk.label(self.login_dialog, text="用户名:").grid(row=2, column=0, padx=5, pady=5)
        self.username_entry = ttk.entry(self.login_dialog)
        self.username_entry.grid(row=2, column=1, padx=5, pady=5)
 
        ttk.button(self.login_dialog, text="登录", command=self.connect_server).grid(row=3, columnspan=2, pady=10)
 
    def connect_server(self):
        server = self.server_entry.get()
        port = self.port_entry.get()
        self.username = self.username_entry.get().strip()
 
        if not self.username:
            messagebox.showerror("错误", "用户名不能为空")
            return
 
        try:
            self.client_socket = socket.socket(socket.af_inet, socket.sock_stream)
            self.client_socket.connect((server, int(port)))
            self.client_socket.send(self.username.encode())
            
            response = self.client_socket.recv(1024).decode()
            if response != "connect_success":
                messagebox.showerror("错误", f"连接失败: {response}")
                return
 
            self.running = true
            self.login_dialog.destroy()
            self.master.title(f"在线聊天室 - {self.username}")
            threading.thread(target=self.receive_messages, daemon=true).start()
 
        except exception as e:
            messagebox.showerror("连接失败", str(e))
            if self.client_socket:
                self.client_socket.close()
 
    def receive_messages(self):
        buffer = ""
        while self.running:
            try:
                data = self.client_socket.recv(4096)
                if not data: break
                
                buffer += data.decode()
                while "\n" in buffer:
                    msg, buffer = buffer.split("\n", 1)
                    if msg.startswith("userlist:"):
                        self.update_user_list(msg[9:].split(','))
                    elif msg.startswith("system:"):
                        self.display_system_msg(msg[7:])
                    else:
                        self.display_message(msg)
            except:
                break
 
    def update_user_list(self, users):
        current = set(self.user_list.get_children())
        online = set(users)
        
        for user in current - online:
            self.user_list.delete(user)
        
        for user in online - current:
            self.user_list.insert("", "end", iid=user, text=user)
 
    def display_message(self, msg):
        self.chat_area.config(state=tk.normal)
        self.chat_area.insert(tk.end, msg + "\n")
        self.chat_area.see(tk.end)
        self.chat_area.config(state=tk.disabled)
 
    def display_system_msg(self, msg):
        self.chat_area.config(state=tk.normal)
        self.chat_area.insert(tk.end, f"【系统】{msg}\n", 'system')
        self.chat_area.see(tk.end)
        self.chat_area.config(state=tk.disabled)
 
    def send_message(self, event=none):
        msg = self.msg_entry.get().strip()
        if msg:
            try:
                self.client_socket.send(f"{msg}\n".encode())
                self.msg_entry.delete(0, tk.end)
                if msg.startswith("@@"):
                    self.display_message(f"[我] 私聊 {msg[2:].split(' ')[0]}：{' '.join(msg.split()[1:])}")
                else:
                    self.display_message(f"[我]：{msg}")
            except exception as e:
                messagebox.showerror("发送失败", str(e))
 
if __name__ == "__main__":
    root = tk.tk()
    app = chatclient(root)
    root.mainloop()