基于Python的毕设选题系统实战：从需求建模到高并发选题冲突处理-开发者社区

背景痛点：抢选题就像春运抢票

高校毕设季，几千名学生同时登录教务系统，热门课题几秒被抢光，冷门课题无人问津。传统做法往往只有一张student_topic表，状态字段status=0/1，并发时大量UPDATE撞车，导致：

超卖：同一课题被 N 个同学“成功”写入，数据库最后只落一条记录，其余全部丢失。
脏读：学生 A 看到“可选”，点击确定瞬间被 B 抢走，前端仍提示“成功”，刷新后却消失。
用户体验差：页面转圈 5s 返回“系统繁忙”，再刷新课题已被抢光。

一句话：没有互斥、没有幂等、没有实时反馈。

技术选型：为什么不是 Django/FastAPI？

维度	Flask	Django	FastAPI
学习/改造成本	低，单文件即可启动	高，ORM+Admin 全家桶	中，依赖 Pydantic 类型体操
生态灵活度	高，自由组合 SQLAlchemy、Redis	中，Django ORM 深度绑定	高，但异步驱动需全链路 async
并发模型	同步+gevent 即可满足 1k QPS	同步	异步
教学场景落地速度	最快，毕设周期 2-3 周	慢	中

结论：教学管理系统业务简单、流量突发、交付周期短，Flask 足够且最省时间。
Redis 作为单线程原子性的内存数据库，天然适合“分布式抢锁”场景，后续横向扩展也无需改代码。

系统架构速览

Nginx 反向代理 + Gunicorn（gevent）
Flask 无状态服务，水平扩容
MySQL 8.0 存储业务数据
Redis 6.2 负责分布式锁、选题热度计数、接口防刷令牌桶

核心实现

1. 数据模型（SQLAlchemy）

class Topic(db.Model): __tablename__ = 'topic' id = db.Column(db.Integer, primary_key=True) title = db.Column(db.String(120)) teacher = db.Column(db.String(40)) quota = db.Column(db.SmallInteger, default=1) # 名额 picked = db.Column(db.SmallInteger, default=0) # 已选 status = db.Column(db.String(10), default='open') # open/close __table_args__ = ( db.Check.SQLOnConflict('quota', 'picked'), # 业务层兜底 )

2. 幂等性设计

利用学生+课题唯一索引，保证同一学生重复点击只产生一条记录。

class Choice(db.Model): __tablename__ = 'choice' id = db.Column(db.Integer, primary_key=True) student_id = db.Column(db.Integer, nullable=False) topic_id = db.Column(db.Integer # 外键 ctime = db.Column(db.DateTime, server_default=func.now()) __table_args__ = ( db.UniqueConstraint('student_id', 'topic_id', name='uk_student_topic'), )

接口层返回相同结果，前端无需额外提示。

3. 分布式锁（Redis Lua 脚本）

import redis, uuid, time r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True) def acquire_lock(key: str, expire: int = 5) -> str: """返回 token，失败返回空字符串""" token = str(uuid.uuid4()) ok = r.set(key, token, nx=True, ex=expire) return token if ok else '' def release_lock(key: str, token: str) -> bool: lua = """ if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end """ return bool(r.eval(lua, 1, key, token))

4. 选题提交接口（含事务隔离）

@bp.post('/choose') def choose(): student_id = g.user.id topic_id = request.json['topic_id'] lock_key = f"lock:topic:{topic_id}" token = acquire_lock(lock_key) if not token: return {'ok': False, 'msg': '系统繁忙，请重试'}, 409 try: # 可重复读，防止幻读 with db.session.begin(): topic = (db.session.query(Topic) .filter_by(id=topic_id) .with_for_update() .first()) if not topic or topic.status != 'open': return {'ok': False, 'msg': '课题已关闭'} if topic.picked >= topic.quota: return {'ok': False, 'msg': '名额已满'} try: db.session.add(Choice(student_id=student_id, topic_id=topic_id)) topic.picked += 1 if topic.picked >= topic.quota: topic.status = 'close' db.session.flush() except IntegrityError: # 唯一索引冲突，幂等返回成功 return {'ok': True, 'msg': '已选过'} finally: release_lock(lock_key, token) return {'ok': True, 'msg': '选题成功'}

关键点：

with_for_update()把行锁与 Redis 分布式锁双保险，即使锁超时仍有数据库兜底。
事务隔离级别REPEATABLE READ（MySQL 默认），避免幻读。
捕获IntegrityError实现幂等，重复点击不报错。

性能与安全

冷启动延迟

Flask 默认懒加载，首次请求 import 全部模型导致 300~500 ms。使用gunicorn --preload预加载，延迟降到 50 ms 内。

防刷机制

接口令牌桶（Redis + Lua）：

def allow(uid: str, rate: int = 5) -> bool: key = f"rate:{uid}" curr = r.incr(key) if curr == 1: r.expire(key, 1) # 1 秒窗口 return curr <= rate

选题接口限流 5 次/秒，超过直接返回 429，保护下游 MySQL。

SQL 注入

SQLAlchemy ORM 已参数化，拒绝原生拼接；额外开启 MySQLsql_mode=STRICT_TRANS_TABLES防隐式转换。

生产避坑指南

锁超时：
- 默认 5 s，接口 RT 99 线 200 ms 内足够；
- 若 GC 或网络抖动导致超时，需配合with_for_update()兜底。
回滚策略：
- 任何异常退出先释放锁，再抛异常，防止死锁。
- 利用try/finally保证锁一定被删掉。
日志追踪：
- 在锁 key 中加入trace_id，通过 ELK 聚合，可快速定位哪一步 RT 过高。
- 记录student_id+topic_id+result，方便审计。
监控：
- Prometheus + Grafana 采集picked/quota比例，提前发现“超卖”风险。
- Redis 内存 >80% 自动扩容，否则set nx失败率飙升。

可扩展方向

多轮志愿：
把Choice表加round字段，定时任务按志愿序+权重撮合，解锁未被命中课题。
热度排行榜：
用 RedisZINCRBY topic:hot 1 <topic_id>，实时展示 Top20，前端 WebSocket 推送。
教师确认：
增加teacher_confirm状态，支持导师“反选”，流程更贴合实际。

写在最后

整个系统 3 周完成，压测 1 k 并发、5 k 选题无超卖，代码行数不到 1 k。把并发冲突拆成“分布式锁 + 数据库行锁 + 唯一索引”三层，层层兜底，既保证安全又留足扩展空间。下一步我准备把热度排行榜做成实时弹幕，让学生像看直播一样刷选题。如果你也做过类似系统，欢迎留言交流更优雅的实现。