简介基于Python和MySQL实现的简易聊天室系统完整源码面向Python网络编程初学者、课程设计与毕业设计开发者结合Socket多线程通信、MySQL数据持久化与图形界面交互完整演示了从登录注册到群聊私聊的桌面聊天工具开发流程。压缩包内共23个文件以8个Python源文件为核心另含5个编译缓存文件、8张界面相关图片、1个功能演示动图和1份README说明文档总大小仅423KB结构清晰便于快速查阅。系统包含用户管理、群聊、私聊、在线用户列表与聊天记录展示等功能用户账号及消息均存放在MySQL中客户端与服务端通过多线程并发处理实时通信。代码按客户端、服务端、数据库操作及界面面板分模块组织配合截图和动图可直观理解运行效果README提供项目说明适合作为网络编程或Python综合应用的课程设计参考。目前已有85人浏览学习对想快速上手聊天室项目并二次开发的读者有一定参考价值。1. 从 Socket 到 MySQL这个简易聊天室系统到底拆了什么如果你写过 Python 脚本却没见过一台机器上的两个进程怎么实时通信这个聊天室是很典型的切入点。它把登录注册、群聊私聊、在线列表和历史记录打包成一个完整的 C/S 项目服务端用原生 Socket 监听端口客户端用 Tkinter 做图形界面MySQL 负责用户和聊天记录的持久化。压缩包里除了 main.py、chat_Server.py、chat_mysql.py 等源码还带着pycache里的 .pyc 文件说明是 Python 3.10 环境实际运行过的。这个项目适合想理解多线程网络编程、准备课程设计或毕设的人也适合那些已经能写爬虫、但没碰过 Socket 编程的开发者。2. Socket 多线程服务端与客户端消息按什么规则流动聊天室的核心是消息的实时转发。服务端一个 socket 绑定端口客户端连上来之后服务端为每个连接单独起一个线程这样一个人发消息其他人不会阻塞等待。压缩包里的 chat_Server.py 正是干这件事的而 chat_client.py 负责上送本地输入并接收服务端转发来的消息。下面按服务端怎么收、消息怎么分类、客户端怎么收三层拆开。2.1 服务端一个连接一个线程用字典管理在线用户Socket 服务端的标准动线是socket()、bind()、listen()、accept()。accept 是阻塞的直到有客户端连入才返回一个连接的 socket。这个简易项目的处理方式很直观accept 一次就启动一个threading.Thread让每个客户端独占一个线程。在线用户则用一个clients字典维护键是用户名值是对应的 socket 连接。import socket import threading import json clients {} # 在线用户: {username: conn} clients_lock threading.Lock() def handle_client(conn, addr): current_user None try: while True: data conn.recv(4096) # 一次性最多读 4096 字节 if not data: break # 客户端主动断开recv 返回空字节串 msg json.loads(data.decode(utf-8)) print(f[{addr}] 收到: {msg}) if msg[type] login: current_user msg[username] with clients_lock: clients[current_user] conn broadcast_online_users() elif msg[type] chat: broadcast(msg[username], msg[content]) elif msg[type] private: send_private(msg[username], msg[to_user], msg[content]) except (ConnectionResetError, json.JSONDecodeError) as e: print(f[客户端断开] {addr}: {e}) finally: with clients_lock: if current_user and clients.get(current_user) conn: del clients[current_user] conn.close() broadcast_online_users() server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((0.0.0.0, 8080)) server.listen(5) print(服务端已启动: 0.0.0.0:8080) while True: conn, addr server.accept() threading.Thread(targethandle_client, args(conn, addr), daemonTrue).start()recv(4096)是单次读取的缓冲上限不代表服务端一次性就能拿到完整消息。TCP 是字节流客户端发的 JSON 可能被拆成几个包也可能几个包粘在一起所以后面要引入分隔符拆包。clients_lock是必要的——多个线程同时del clients[user]或遍历clients时不加锁容易抛RuntimeError: dictionary changed size during iteration。SO_REUSEADDR让服务端重启后能立刻重新绑定同一个端口否则要等系统释放大约几十秒到几分钟。daemonTrue也很关键服务端主线程在accept死循环里CtrlC 退出时daemon 线程不会阻塞进程退出。如果这里不设 daemon你关掉服务端窗口还得一个一个手动杀 Python 进程。2.2 消息协议用 JSON 字段区分群聊、私聊和系统消息多个客户端同时在线时服务端必须知道一条消息是发给所有人还是某个特定的人。这个项目没用 MQTT 或 WebSocket 这类重型方案直接在 TCP 之上定义了一个文本协议每条消息是 JSON 字符串末尾加\n作为边界。字段设计很直白字段类型说明typestrlogin / chat / private / users / systemusernamestr发送者用户名to_userstr私聊接收者群聊时为空字符串contentstr消息正文timestampstr客户端本地时间仅供界面展示一个完整的转发逻辑大概长这样def broadcast(sender, content): target_conns [] with clients_lock: for name, conn in clients.items(): if name ! sender: target_conns.append(conn) obj json.dumps({ type: chat, username: sender, to_user: , content: content }, ensure_asciiFalse) \n for conn in target_conns: try: conn.sendall(obj.encode(utf-8)) except OSError: pass def send_private(sender, to_user, content): with clients_lock: target clients.get(to_user) if not target: return {ok: False, msg: f用户 {to_user} 不在线} obj json.dumps({ type: private, username: sender, to_user: to_user, content: content }, ensure_asciiFalse) \n target.sendall(obj.encode(utf-8)) return {ok: True}json.dumps里的ensure_asciiFalse是为了保留中文原文否则中文会被转成\uXXXX传输体积变大且不方便调试。每条消息末尾追加\n是给接收端切分完整消息用的。sendall会确保一次性发完整个字节串不像send可能只发一半。私聊目标不在线时必须回给发送者一个失败提示这是简易聊天室很容易漏掉的地方——如果直接把消息写进数据库而不提示发送者会以为对方已收到。2.3 客户端一个线程收消息一个线程发命令客户端的职责分成两路UI 主线程等用户点击按钮发送另一个独立线程循环接收服务端推送。如果接收线程在拿到消息后直接操作 Tkinter 控件界面会随机崩溃或卡死所以正确做法是把数据丢进queue.Queue让主线程轮询处理。import socket import json import threading import queue class ChatClient: def __init__(self, host, port): self.sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.connect((host, port)) self.recv_queue queue.Queue() def start_recv(self): def _recv(): buf while True: try: data self.sock.recv(4096).decode(utf-8) if not data: break buf data while \n in buf: line, buf buf.split(\n, 1) # 按换行切出一条完整消息 if line.strip(): self.recv_queue.put(json.loads(line)) except (OSError, json.JSONDecodeError): break threading.Thread(target_recv, daemonTrue).start() def send_message(self, msg): raw json.dumps(msg, ensure_asciiFalse) \n self.sock.sendall(raw.encode(utf-8))这里最容易被忽略的是buf变量。TCP 收包可能一次收到 1.5 条消息也可能半条都不到所以必须先把数据拼进缓冲区再按\n切分。用split(\n, 1)能正确处理一条长消息里包含换行——如果消息正文本身允许带换行就不能用裸\n做分隔符要改成\r\n或自定义长度头。3. MySQL 用户表与聊天记录认证和持久化怎么做压缩包里单独放了一个 chat_mysql.py这个模块把 MySQL 的所有操作集中起来了登录、注册、写聊天记录、读历史记录都从这一层走。下面先说数据库连接怎么封装再讲注册登录的密码处理最后看聊天记录怎么落库和加载。3.1 chat_mysql.py数据库连接集中封装直接用 pymysql 连接数据库时每个函数都重写一遍 connect 会很啰嗦。这个项目采用的是每次操作新建连接用完关闭的短连接模式因为聊天室并发量不大长连接反而容易把 MySQL 的连接数撑满。import pymysql DB_CONFIG { host: localhost, port: 3306, user: root, password: 123456, database: chat_room, charset: utf8mb4, } def get_conn(): return pymysql.connect( **DB_CONFIG, cursorclasspymysql.cursors.DictCursor )charset建议用utf8mb4而不是utf8因为后者存不了 emoji用户昵称或消息里只要有一个表情整条 INSERT 就会报Incorrect string value。DictCursor让查询结果变成字典而不是元组写代码时用row[username]比row[0]可读性好很多。生产环境一般用dbutils.pooled_db连接池或 SQLAlchemy 管理会话但课程设计级别短连接完全够用。需要注意的是DB_CONFIG里的密码必须改成你本机 MySQL 的真实密码否则登录面板一提交就会抛Access denied for user rootlocalhost。3.2 注册与登录密码不能明文落库源码里没有附带建表 SQL 的话一般 README 会给出初始化脚本或者 chat_mysql.py 启动时自己执行建表。用户表结构建议这样设计字段类型约束idINT自增主键usernameVARCHAR(50)唯一索引passwordVARCHAR(128)存储加盐哈希不是明文created_atTIMESTAMP默认 CURRENT_TIMESTAMP密码不能直接存。项目体量再小也不能把用户输入的字符串原样写进password列否则数据库一旦泄露所有账号直接暴露。一个足够用的方案是加盐哈希import hashlib import os def hash_password(password: str) - str: # 生成 16 字节随机盐转成 32 位十六进制字符串 salt os.urandom(16).hex() digest hashlib.sha256( (salt password).encode(utf-8) ).hexdigest() return f{salt}:{digest} def verify_password(password: str, stored: str) - bool: salt, digest stored.split(:) return hashlib.sha256( (salt password).encode(utf-8) ).hexdigest() digestos.urandom(16)每次生成不同的随机盐所以同一个密码注册两次得到的哈希串完全不同。验证时从库里取出salt:digest用同样的方式重新计算再比较。真正的生产环境应该用 bcrypt 或 argon2因为 sha256 对 GPU 破解并不够慢但课程设计展示思路已经足够。注册和登录的 SQL 必须用参数化查询不要拼字符串def register(username, password): conn get_conn() try: with conn.cursor() as cur: cur.execute( INSERT INTO users (username, password) VALUES (%s, %s), (username, hash_password(password)) ) conn.commit() return True except pymysql.err.IntegrityError: # username 唯一约束重复注册会直接抛异常 return False finally: conn.close() def login(username, password): conn get_conn() try: with conn.cursor() as cur: cur.execute( SELECT password FROM users WHERE username %s, (username,) ) row cur.fetchone() if row and verify_password(password, row[password]): return True return False finally: conn.close()cur.execute使用%s占位符参数通过第二个参数传进去这样输入内容再特殊也不会改变 SQL 结构。注册时捕获IntegrityError正好利用数据库的唯一索引拦住重名账号。登录时不要区分用户不存在和密码错误两种情况统一返回失败避免暴露账号是否已注册。3.3 聊天记录写入与历史加载什么时候查库服务端收到一条群聊或私聊消息后除了转发给在线客户端还要把消息写进 messages 表。最简单的做法是在当前处理线程里直接 INSERT消息量不大时不会有性能问题。历史加载的 SQL 一般长这样-- 群聊取最近 50 条 SELECT sender, receiver, content, sent_at FROM messages WHERE receiver IS NULL OR receiver ORDER BY id DESC LIMIT 50; -- 私聊取两人之间的往来消息 SELECT sender, receiver, content, sent_at FROM messages WHERE (sender A AND receiver B) OR (sender B AND receiver A) ORDER BY id ASC;群聊记录的 receiver 存空字符串私聊记录用双方用户名组合查询。如果私聊历史用LIKE %A%B%的方式查会混入其他无关消息所以必须用两个 OR 条件精确匹配双向记录。为了让私聊历史查询有索引可用messages 表最好加一个INDEX idx_private (sender, receiver)。4. Tkinter 登录与主面板GUI 多线程的安全更新方式这个项目的图形界面拆成了三个文件chat_login_panel.py、chat_register_panel.py、chat_main_panel.py再加上 main.py 做入口。Tkinter 不适合在子线程里直接操作控件所以项目用回调函数传递事件socket 线程的数据统一进队列再由 UI 主线程轮询。下面按界面流转顺序拆。4.1 三个面板怎么组织main.py 做状态切换main.py 创建 Tk 根窗口后根据当前状态显示登录页或聊天页。登录成功后把用户名、客户端连接对象、队列传给聊天面板。三个面板的职责划分如下文件核心控件动作chat_login_panel.pyEntry, Button, Label输入账号密码点击登录或跳转注册chat_register_panel.pyEntry, Button输入新账号密码提交注册chat_main_panel.pyText, Listbox, Entry, Button显示聊天记录、在线用户、发送消息Tkinter 通用的启动流程import tkinter as tk from chat_login_panel import LoginPanel def run_app(): root tk.Tk() root.title(简易聊天室登录) root.geometry(360x280) # on_success 回调在登录成功后替换整个窗口内容 LoginPanel(root, on_successgo_chat) root.mainloop() if __name__ __main__: run_app()登录成功后的go_chat回调里要清空当前窗口的所有控件再创建聊天面板对象。不要一边开着登录窗口一边又tk.Toplevel新建聊天窗那样关掉任意一个窗口都会让程序进退两难。用同一个 root 切换 Frame 是更干净的做法。4.2 聊天窗口用 queue 把 socket 数据搬到 UI 线程客户端接收线程拿到服务端广播后不能直接执行text_area.insert(...)。Tkinter 控件的方法调用没有加锁子线程更新界面时主线程可能正好在重绘偶发崩溃极难排查。项目里正确的做法是接收线程只会做一件事把解析好的消息放进queue.Queue。主线程用root.after定时轮询import queue import tkinter as tk class ChatMainPanel: def __init__(self, root, client, username): self.root root self.client client self.username username self.msg_queue queue.Queue() self.text_area tk.Text(root, height20, width60) self.text_area.pack(sidetk.LEFT) self.user_list tk.Listbox(root, width15, height20) self.user_list.pack(sidetk.RIGHT, filltk.Y) self.input_entry tk.Entry(root, width45) self.input_entry.pack(sidetk.LEFT) send_btn tk.Button(root, text发送, commandself.send_chat) send_btn.pack(sidetk.RIGHT) self.client.start_recv() # 子线程往 msg_queue 塞消息 self.root.after(100, self.poll_queue) def poll_queue(self): try: while True: msg self.msg_queue.get_nowait() self._show_msg(msg) # 仅在主线程更新控件 except queue.Empty: pass self.root.after(100, self.poll_queue) # 每 100ms 再查一次 def send_chat(self): content self.input_entry.get().strip() if content: self.client.send_message({ type: chat, username: self.username, to_user: , content: content, }) self.input_entry.delete(0, tk.END) def _show_msg(self, msg): # 根据 msg[type] 决定插入文本的颜色和格式 if msg[type] chat: self.text_area.insert(tk.END, f{msg[username]}: {msg[content]}\n) elif msg[type] private: self.text_area.insert(tk.END, f[私聊] {msg[username]}: {msg[content]}\n)root.after(100, self.poll_queue)的意思是让 Tkinter 主循环每 100 毫秒调用一次poll_queue。这个间隔不能太大否则消息显示有明显延迟但也没必要低于 50 毫秒否则会挤占界面重绘。发送消息时直接在 UI 线程里调sendall如果网络卡顿界面会短暂无响应要想完全流畅发送动作也应该丢进线程池不过这个项目的体量下不需要。4.3 在线列表刷新与私聊弹窗复用同一套协议在线用户列表靠服务端下发typeusers的系统消息维护。客户端收到后把 Listbox 清空再重新插入。私聊弹窗在用户双击列表项时创建里面单独放一个 Text 和 Entry并记录当前的to_user。def refresh_users(self, user_list): self.user_list.delete(0, tk.END) for name in user_list: if name ! self.username: self.user_list.insert(tk.END, name) def open_private_chat(self, event): selection self.user_list.curselection() if not selection: return peer self.user_list.get(selection[0]) PrivateChatWindow(self.root, self.client, self.username, peer)双击列表项会触发 Tkinter 的ListboxSelect事件但注意这个事件在点击时可能触发多次所以要通过curselection()返回值做一次判空。刷新用户列表时把自己过滤掉否则用户会看到列表里又出现自己的名字点击自己发私聊会产生一种奇怪的体验。5. 跑通项目与排错从 pip 安装到首次登陆把这个项目跑起来比预想中更依赖环境顺序。压缩包里的 .pyc 文件是 Python 3.10 生成的如果你的解释器版本不同最好先删掉pycache里的*.cpython-310.pyc否则会看到interpreter version mismatch之类的坏文件警告。然后按下面步骤走。先确认 Python 和 MySQL 都装好了安装 pymysqlpip install pymysql如果机器上有多个 Python用python -m pip install pymysql保证装到当前解释器对应的目录。接下来启动 MySQL 并初始化库表CREATE DATABASE IF NOT EXISTS chat_room DEFAULT CHARACTER SET utf8mb4; USE chat_room; CREATE TABLE users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50) NOT NULL UNIQUE, password VARCHAR(128) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE messages ( id INT AUTO_INCREMENT PRIMARY KEY, sender VARCHAR(50) NOT NULL, receiver VARCHAR(50) DEFAULT NULL, content TEXT, sent_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_private (sender, receiver) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;messages表上的idx_private索引就是给 3.3 节那条私聊 OR 查询用的否则在线人数一多私聊历史加载会越来越慢。初始化后修改 chat_mysql.py 里的DB_CONFIG密码先启动chat_Server.py再运行main.py注册一个账号试试。运行中常见的报错和定位顺序OperationalError: (2003, Cant connect to MySQL server)MySQL 服务没启动。Windows 在服务管理器里看 MySQLLinux 用systemctl status mysql确认状态。Access denied for user rootlocalhost数据库密码与DB_CONFIG不一致需要修改配置文件里的 password。Address already in use上一次服务端进程没退出。命令行执行netstat -ano | findstr :8080找到 PID 杀进程或者把 bind 端口改掉。登录后立刻闪退客户端接收线程异常没被捕获。先用普通 Python 脚本跑 chat_client.py看终端有没有 traceback如果中文乱码检查 MySQL 库、表、连接三处字符集是否都是utf8mb4。如果怀疑是消息协议的问题最好的办法是不经 GUI 直接验证服务端用telnet 127.0.0.1 8080手动连接粘贴一行 JSON 字符串观察服务端回显。这样能把服务端逻辑错误和客户端界面错误快速分开而不是对着 Tkinter 的报错猜来猜去。本文还有配套的精品资源点击获取