简介：基于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(target=handle_client, args=(conn, addr), daemon=True).start()recv(4096)是单次读取的缓冲上限，不代表服务端一次性就能拿到完整消息。TCP 是字节流，客户端发的 JSON 可能被拆成几个包，也可能几个包粘在一起，所以后面要引入分隔符拆包。clients_lock是必要的——多个线程同时del clients[user]或遍历clients时，不加锁容易抛RuntimeError: dictionary changed size during iteration。SO_REUSEADDR让服务端重启后能立刻重新绑定同一个端口，否则要等系统释放，大约几十秒到几分钟。
daemon=True也很关键：服务端主线程在accept死循环里，Ctrl+C 退出时，daemon 线程不会阻塞进程退出。如果这里不设 daemon，你关掉服务端窗口还得一个一个手动杀 Python 进程。
2.2 消息协议：用 JSON 字段区分群聊、私聊和系统消息
多个客户端同时在线时，服务端必须知道一条消息是发给所有人还是某个特定的人。这个项目没用 MQTT 或 WebSocket 这类重型方案，直接在 TCP 之上定义了一个文本协议：每条消息是 JSON 字符串，末尾加\n作为边界。字段设计很直白：
| 字段 | 类型 | 说明 |
|---|---|---|
| type | str | login / chat / private / users / system |
| username | str | 发送者用户名 |
| to_user | str | 私聊接收者，群聊时为空字符串 |
| content | str | 消息正文 |
| timestamp | str | 客户端本地时间，仅供界面展示 |
一个完整的转发逻辑大概长这样：
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_ascii=False) + '\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_ascii=False) + '\n' target.sendall(obj.encode('utf-8')) return {'ok': True}json.dumps里的ensure_ascii=False是为了保留中文原文，否则中文会被转成\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, daemon=True).start() def send_message(self, msg): raw = json.dumps(msg, ensure_ascii=False) + '\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, cursorclass=pymysql.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 'root'@'localhost'。
3.2 注册与登录：密码不能明文落库
源码里没有附带建表 SQL 的话，一般 README 会给出初始化脚本，或者 chat_mysql.py 启动时自己执行建表。用户表结构建议这样设计：
| 字段 | 类型 | 约束 |
|---|---|---|
| id | INT | 自增主键 |
| username | VARCHAR(50) | 唯一索引 |
| password | VARCHAR(128) | 存储加盐哈希，不是明文 |
| created_at | TIMESTAMP | 默认 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.py | Entry, Button, Label | 输入账号密码，点击登录或跳转注册 |
| chat_register_panel.py | Entry, Button | 输入新账号密码，提交注册 |
| chat_main_panel.py | Text, 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_success=go_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, height=20, width=60) self.text_area.pack(side=tk.LEFT) self.user_list = tk.Listbox(root, width=15, height=20) self.user_list.pack(side=tk.RIGHT, fill=tk.Y) self.input_entry = tk.Entry(root, width=45) self.input_entry.pack(side=tk.LEFT) send_btn = tk.Button(root, text="发送", command=self.send_chat) send_btn.pack(side=tk.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 在线列表刷新与私聊弹窗：复用同一套协议
在线用户列表靠服务端下发type=users的系统消息维护。客户端收到后，把 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 都装好了，安装 pymysql：
pip 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 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; 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) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;messages表上的idx_private索引就是给 3.3 节那条私聊 OR 查询用的，否则在线人数一多，私聊历史加载会越来越慢。初始化后修改 chat_mysql.py 里的DB_CONFIG密码，先启动chat_Server.py，再运行main.py，注册一个账号试试。
运行中常见的报错和定位顺序：
OperationalError: (2003, "Can't connect to MySQL server")：MySQL 服务没启动。Windows 在服务管理器里看 MySQL，Linux 用systemctl status mysql确认状态。Access denied for user 'root'@'localhost'：数据库密码与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 的报错猜来猜去。
本文还有配套的精品资源，点击获取
