基于Python實(shí)現(xiàn)多人聊天室的示例代碼

更新時(shí)間：2025年02月20日 10:23:31 作者：席子哥哥的代碼庫

這篇文章主要為大家詳細(xì)介紹了如何基于Python實(shí)現(xiàn)多人聊天室功能,文中的示例代碼講解詳細(xì),具有一定的借鑒價(jià)值,有需要的小伙伴可以參考下

本文主要為大家介紹一下如何實(shí)現(xiàn)一個(gè)多人聊天室（server+client），感興趣的小伙伴可以了解下

效果圖

通過本地服務(wù)器以用戶名登錄

實(shí)現(xiàn)關(guān)鍵代碼

支持群聊和私聊

sever端代碼：

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"?? 服務(wù)端已啟動在 {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} 進(jìn)入聊天室")
            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)
        
        # 添加發(fā)送按鈕
        send_btn = ttk.Button(input_frame, text="發(fā)送", 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="服務(wù)器地址:").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"【系統(tǒng)】{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("發(fā)送失敗", str(e))
 
if __name__ == "__main__":
    root = tk.Tk()
    app = ChatClient(root)
    root.mainloop()

以上就是基于Python實(shí)現(xiàn)多人聊天室的示例代碼的詳細(xì)內(nèi)容，更多關(guān)于Python多人聊天室的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: