Python實現網絡圖形化界面多人聊天室 - Windows
項目名稱：網絡多人聊天室圖形界面版本
項目思路：
server.py
服務端文件，主進程中，創建圖形化界面，詢問地址（主機名，端口），點擊開始進入聊天室。
創建子進程，開始網絡連接，使用select.select循環接收客戶端連接請求，使用timeout不斷檢查與主進程間隊列（multiprocessing.Queues）的情況
client.py
客戶端文件，主進程中，創建圖形化界面，詢問地址（主機名，端口），點擊開始以連接到客戶端。
創建子進程，開始網絡連接。
settings.py
默認設置
# settings.py # 默認設置 HOST = "127.0.0.1" PORT = 5555 ADDR = HOST, PORT def center(root, width=300, height=150): # 設置窗口居中 screenWidth = root.winfo_screenwidth() screenHeight = root.winfo_screenheight() x = (screenWidth - width) / 2 y = (screenHeight - height) / 2 root.geometry("%dx%d+%d+%d" % (width, height, x, y))
# server.py # 服務端代碼 from time import ctime from multiprocessing import Process, Queue from select import select from socket import * from settings import * from tkinter import * from tkinter.scrolledtext import ScrolledText from tkinter import messagebox def main_gui(): # 主窗口 root = Tk() # 設置窗口居中 center(root) # 設置窗口其他屬性 root.title("多人聊天室主窗口") root.resizable(0, 0) root.configure(bg="white") # root.iconbitmap("python.ico") # 添加主機名（HOST）以及端口號（PORT）等輸入框 pad = 10 Label(root, text="主機名（Host）：").grid(row=0, column=0, padx=pad, pady=pad) ent_host = Entry(root) ent_host.insert(0, HOST) ent_host.grid(row=0, column=1, padx=pad, pady=pad) Label(root, text="端口號（Port）：").grid(row=1, column=0, padx=pad, pady=pad) ent_port = Entry(root) ent_port.insert(0, PORT) ent_port.grid(row=1, column=1, padx=pad, pady=pad) # 組件列表 widgets = { "ent_host": ent_host, "ent_port": ent_port } # 添加確認按鈕 btn_cfm = Button(root, text="新建網絡聊天室", command=lambda:validate(root, widgets)) btn_cfm.grid(rowspan=2, columnspan=2, padx=pad, pady=pad) # 綁定事件 root.bind("<Return>", lambda event:validate(root, widgets)) # 主循環事件 root.mainloop() def validate(root, widgets): # 確認按鈕事件，檢查是否輸入有誤 host, port = widgets["ent_host"].get(), widgets["ent_port"].get() # 如果端口號不是純數字 try: port = int(port) except: messagebox.showerror("錯誤", "端口號必須為數字！") return # 彈出錯誤窗口 if not host or not port: messagebox.showerror("錯誤", "主機名或端口不可為空！") return # 有效地址 addr = (host, port) # 檢查是否套接字成功 try: server = socket(AF_INET, SOCK_STREAM) server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) server.bind(addr) server.listen(10) except Exception as e: messagebox.showerror("錯誤", f"無法在{addr}上生成套接字！") print(e) return else: # 生成兩個隊列 # queue負責在子進程中循環get主進程的信息 # queueu負責在父進程中循環get子進程的消息 queue = Queue() queueu = Queue() # 生成子進程 process = Process(target=inet, args=(server, queue, queueu)) process.daemon = True process.start() # 創建聊天室頁面 chatroom_gui(root, queue, queueu) def chatroom_gui(r, queue, queueu): # 聊天室頁面 r.destroy() root = Tk() # 設置窗口居中 center(root, 500, 500) # 最小化窗口 root.minsize(350, 350) # 菜單欄 menubar = Menu(root) menubar.add_command(label="新建", command=None) menubar.add_command(label="信息", command=None) menubar.add_command(label="退出", command=root.destroy) root.config(menu=menubar) # 文本框 text = ScrolledText(root) text.pack(fill=BOTH) # 輸入框 ent = Entry(root, bg='gray', bd=3) ent.pack(fill=BOTH, side=BOTTOM) ent.focus_set() # 綁定事件 root.bind("<Return>", lambda event:send(ent, queue, text)) # 設置窗口其他屬性 root.title("多人聊天室 - 管理員") root.configure(bg="white") root.iconbitmap("python.ico") # 主循環函數 root.after(1000, recv, root, queueu, text) def recv(root, queueu, text): if not queueu.empty(): data = queueu.get() text.insert(END, data) root.after(1000, recv, root, queueu, text) def send(ent, queue, text): now = ":".join(ctime().split()[3].split(":")) data = "[" + now + "] " + "管理員：" + ent.get() + "\n" queue.put(data) text.insert(END, data) ent.delete(0, END) def inet(server, queue, queueu): # 子進程 rlist = [server] wlist = [] xlist = [] while True: # 接收隊列信息 if not queue.empty(): data = queue.get() for conn in rlist: if conn is not server: conn.send(bytes(data, "UTF-8")) rs, ws, xs = select(rlist, wlist, xlist, 1) for r in rs: if r is server: conn, addr = r.accept() rlist.append(conn) else: try: data = r.recv(1024) except: rlist.remove(r) else: queueu.put(data.decode()) for conn in rlist: if conn is not server: conn.send(data) def main(): # 主進程 main_gui() if __name__ == "__main__": main()
# client.py # 客戶端代碼 import sys from time import sleep, ctime from multiprocessing import Process, Queue from socket import * from settings import * from tkinter import * from tkinter.scrolledtext import ScrolledText from tkinter import messagebox def main_gui(): # 主窗口 root = Tk() # 設置窗口居中 center(root, 300, 200) # 設置窗口其他屬性 root.title("多人聊天室主窗口") root.resizable(0, 0) root.configure(bg="white") root.iconbitmap("python.ico") # 添加主機名（HOST）以及端口號（PORT）等輸入框 pad = 10 Label(root, text="主機名（Host）：").grid(row=0, column=0, padx=pad, pady=pad) ent_host = Entry(root) ent_host.insert(0, HOST) ent_host.grid(row=0, column=1, padx=pad, pady=pad) Label(root, text="端口號（Port）：").grid(row=1, column=0, padx=pad, pady=pad) ent_port = Entry(root) ent_port.insert(0, PORT) ent_port.grid(row=1, column=1, padx=pad, pady=pad) Label(root, text="用戶名（User）：").grid(row=2, column=0, padx=pad, pady=pad) ent_user = Entry(root) ent_user.grid(row=2, column=1, padx=pad, pady=pad) ent_user.focus_set() # 組件列表 widgets = { "ent_host": ent_host, "ent_port": ent_port, "ent_user": ent_user } # 添加確認按鈕 btn_cfm = Button(root, text="加入目標聊天室", command=lambda:validate(root, widgets)) btn_cfm.grid(rowspan=2, columnspan=2, padx=pad, pady=pad) # 綁定事件 root.bind("<Return>", lambda event:validate(root, widgets)) # 主循環事件 root.mainloop() def validate(root, widgets): # 確認按鈕事件，檢查是否輸入有誤 host, port, user = widgets["ent_host"].get(), widgets["ent_port"].get(), widgets["ent_user"].get() # 如果端口號不是純數字 try: port = int(port) except: messagebox.showerror("錯誤", "端口號必須為數字！") return # 彈出錯誤窗口 if not host or not port or not user: messagebox.showerror("錯誤", "主機名或端口或用戶名不可為空！") return # 有效地址 addr = (host, port) # 檢查是否套接字成功 try: client = socket(AF_INET, SOCK_STREAM) client.connect(addr) except Exception as e: messagebox.showerror("錯誤", f"無法在{addr}上生成套接字！") print(e) return else: # 生成兩個隊列 # queue負責在子進程中循環get主進程的信息 # queueu負責在父進程中循環get子進程的消息 queue = Queue() queueu = Queue() # 發送成功建立連接信息 client.send(bytes(f"{user}進入了聊天室。\n", "UTF-8")) # 生成子進程 process = Process(target=inet, args=(client, queue, queueu, user)) process.daemon = True process.start() # 創建聊天室頁面 chatroom_gui(root, queue, queueu, user) def chatroom_gui(r, queue, queueu, user): # 聊天室頁面 r.destroy() root = Tk() # 設置窗口居中 center(root, 500, 500) # 最小化窗口 root.minsize(350, 350) # 菜單欄 menubar = Menu(root) menubar.add_command(label="信息", command=None) menubar.add_command(label="退出", command=root.destroy) root.config(menu=menubar) # 文本框 text = ScrolledText(root) text.pack(fill=BOTH) # 輸入框 ent = Entry(root, bg='gray', bd=3) ent.pack(fill=BOTH, side=BOTTOM) ent.focus_set() # 綁定事件 root.bind("<Return>", lambda event:send(ent, queue, user)) # 設置窗口其他屬性 root.title(f"多人聊天室 - {user}") root.configure(bg="white") # root.iconbitmap("python.ico") # 設置退出方法 root.protocol("WM_DELETE_WINDOW", lambda: exit(root, queue, user)) # 主循環函數 root.after(1000, recv, root, queueu, text) def exit(root, queue, user): data = user + "退出了聊天室。\n" queue.put(data) sleep(1) root.destroy() sys.exit(0) def recv(root, queueu, text): if not queueu.empty(): data = queueu.get() if data == 404: messagebox.showerror("錯誤", "服務端關閉了連接！") sys.exit(0) text.insert(END, data) root.after(1000, recv, root, queueu, text) def send(ent, queue, user): now = ":".join(ctime().split()[3].split(":")) data = "[" + now + "] " + user + "：" + ent.get() + "\n" queue.put(data) ent.delete(0, END) def inet(client, queue, queueu, user): # 子進程 client.setblocking(0) while True: if not queue.empty(): data = queue.get() if data: data = bytes(data, "UTF-8") client.send(data) try: data = client.recv(1024) except BlockingIOError as e: continue except: queueu.put(404) else: data = data.decode() queueu.put(data) def main(): # 主進程 main_gui() if __name__ == "__main__": main()
運行截圖：
