拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Python Flask在线选课系统开发实战:并发控制与数据库设计

Python Flask在线选课系统开发实战:并发控制与数据库设计

1. 从零搭建在线选课系统:需求分析与技术选型

作为一个常年泡在校园信息化项目里的开发者,我接手过不少类似"在线选课系统"的活儿。这个标题里"基于python"很明确,技术栈主Python,而"hx4616"大概率是课程设计或毕设的编号。说实话,这类系统的业务复杂度不算高,但它麻雀虽小五脏俱全,涉及用户认证、选课冲突检测、并发处理、权限控制这些经典问题,非常适合拿来练手,也适合作为“Python全栈入门到实战”的样板项目。

本文就基于我当时做的一个选课系统,从需求设计、数据库建模、核心代码实现到部署排查,把完整链路拆开讲清楚。不管你是要应付课程设计,还是想在公司内部做个类似的报名/预约系统,这套思路都能直接复用。

1.1 先弄清楚系统到底要解决什么问题

在线选课系统听起来好像就是“学生选课、老师开课”,但真正落地时,需求往往是这样的:

  • 学生端要能浏览课程列表、查看课程余量、提交选课、退课、查看个人课表。
  • 教师(或管理员)端要能登录后创建课程、设置容量和时间、查看选课名单。
  • 管理后台还要处理选课时间窗口的控制,比如“第1-2周开放选课,之后自动关闭”。
  • 系统必须防止超选,也就是当只剩1个名额时,两个人同时点“选课”,只能有一个人成功。

这最后一条是最容易踩坑的地方。很多人用Flask或者Django做CRUD时,直观地就写一个if course.students < capacity: insert,结果一到真实并发场景就超卖。我在代码实现部分会专门讲这个问题怎么解决。

1.2 技术选型:为什么是Flask而不是Django

用Python做这类系统,最主流的两个框架就是Flask和Django。我这次选的是Flask,原因很直接:

  • 项目规模不大,业务逻辑集中,不需要Django自带的Admin后台、ORM全功能套件,Flask更轻、更容易让初学者理清请求处理流程。
  • 选课系统的核心是“接口逻辑 + 数据库事务”,Flask + SQLAlchemy + SQLite/MySQL的组合足够,且代码量更少。
  • Flask的蓝图(Blueprint)机制适合按功能拆模块,比如auth、course、admin,结构清晰,后续扩展也方便。

当然,如果你更习惯Django的“全家桶”风格,或者需要自带的用户认证后台,用Django也完全可以,设计思路是一样的。我这里以Flask为例,但所有SQL和业务逻辑的讲解,换到Django上一样成立。

2. 系统架构设计:三个核心模块的拆解

在线选课系统可以拆成三个核心模块:认证与权限模块、课程管理模块、选课业务模块。这三个模块的边界必须清晰,否则写着写着就成了一锅粥。

2.1 认证与权限模块:角色是如何控制访问的

系统里有三种角色:管理员、教师、学生。我的做法是建立一张users表,里面加一个role字段,用整数区分角色,而不是为每个角色建一张单独的用户表。

class User(db.Model): __tablename__ = 'users' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(64), unique=True, nullable=False) password_hash = db.Column(db.String(128), nullable=False) role = db.Column(db.Integer, default=2) # 0-管理员 1-教师 2-学生

权限控制上,我用Flask的装饰器来做:

from functools import wraps from flask import session, jsonify def role_required(role): def decorator(f): @wraps(f) def wrapper(*args, **kwargs): user = get_current_user() if not user: return jsonify({'code': 401, 'msg': '未登录'}), 401 if user.role != role: return jsonify({'code': 403, 'msg': '无权访问'}), 403 return f(*args, **kwargs) return wrapper return decorator

这里有个细节:不要用布尔字段判断“是否管理员”,因为系统一旦要扩展角色(比如加一个“助教”),布尔字段就不好扩展了。用整数角色值,加一个常量枚举,后续维护就舒服很多。

2.2 课程管理模块:教师创建课程的业务规则

教师创建课程时,除了课程名、学分、上课时间这些基本信息,还需要绑定一个“选课容量”。这个容量是选课冲突判断的关键数据源。

我的课程表设计如下:

class Course(db.Model): __tablename__ = 'courses' id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(128), nullable=False) teacher_id = db.Column(db.Integer, db.ForeignKey('users.id')) capacity = db.Column(db.Integer, default=30) selected_count = db.Column(db.Integer, default=0) schedule = db.Column(db.String(64)) # 如 "周一 3-4节" description = db.Column(db.Text)

注意,我这里用一个selected_count字段记录已选人数。很多人在设计表时喜欢在查询时用db.session.query(Selection).filter_by(course_id=...).count()来实时统计人数,这样在数据量小的时候没问题,但一旦并发高,count查询既慢又容易产生竞态条件。用一个冗余计数字段,配合事务操作,才能保证选课不超卖。

2.3 选课业务模块:事务与锁的正确用法

选课的核心逻辑思政课就三件事:判断是否冲突、扣减名额、记录选课关系。听起来简单,但并发环境下必须保证原子性。

我用的方案是基于数据库行锁的乐观并发控制,也就是SQLAlchemy的with_for_update():

from sqlalchemy import func def select_course(course_id, student_id): course = Course.query.filter_by(id=course_id).with_for_update().first() if not course: return {'code': 404, 'msg': '课程不存在'} if course.selected_count >= course.capacity: return {'code': 400, 'msg': '课程已满'} exists = Selection.query.filter_by(course_id=course_id, student_id=student_id).first() if exists: return {'code': 400, 'msg': '不可重复选课'} course.selected_count += 1 db.session.add(Selection(course_id=course_id, student_id=student_id)) db.session.commit() return {'code': 200, 'msg': '选课成功'}

这里with_for_update()的作用是对这条课程记录加行级锁,另一个请求来选同一门课时,会阻塞等待前一个事务提交或回滚。这样就不会出现两个请求同时读到selected_count=29,然后一起加1的超卖问题。

要注意的是,行锁必须放在事务里才有意义。SQLAlchemy中默认session在第一次查询时开启事务,所以只要你不调用db.session.commit(),锁一直持有。这也是为什么我明确把“查询课程、判断容量、插入记录、提交事务”放在同一个函数里,不能拆成多个独立的session调用。

3. 数据库设计:五张核心表的关联关系

这类系统的数据模型不复杂,但表之间的关联关系需要提前理清,否则后期会频繁改动表结构。

3.1 表结构全景图

整个系统我用了五张核心表:

表名用途关键字段
users用户表id, username, password_hash, role
courses课程表id, name, teacher_id, capacity, selected_count, schedule
selections选课记录表id, student_id, course_id, created_at
announcements公告表id, title, content, created_at
logs操作日志表id, user_id, action, detail, created_at

selections表是选课系统的核心关联表,它连接了学生和课程,是多对多关系的中间表。

class Selection(db.Model): __tablename__ = 'selections' id = db.Column(db.Integer, primary_key=True) student_id = db.Column(db.Integer, db.ForeignKey('users.id')) course_id = db.Column(db.Integer, db.ForeignKey('courses.id')) created_at = db.Column(db.DateTime, default=datetime.utcnow)

我在设计选课记录时加了created_at时间戳,看起来只是多一个字段,但实际派上了很大用场:一是可以作为选课时间排序的依据,二是以后如果要做“选课时间窗口限制”,直接拿这个时间字段和窗口配置比对就行。

3.2 为什么要用关联表而不是在学生表里存课程列表

初学者常犯的一个错误是在users表里加一个courses字段,用逗号分隔存课程ID,比如“3,5,12”。这种设计的坏处非常多:查某个学生的选课列表要拆分字符串,统计课程人数非常麻烦,改退课时更新字符串会出大把Bug。

所以务必使用关联表。在学生选课后,往selections里插入一条记录;退课时删除记录;查“我选了哪些课”时联表查询。这就是关系型数据库的“三范式”思想,虽然不用背理论,但这个场景下关联表的优势是碾压式的。

3.3 数据库迁移工具的使用

Flask + SQLAlchemy 项目中,我强烈建议你从一开始就引入Flask-Migrate来做数据表结构的迁移管理。比如你在开发时加了announcements表,如果直接用db.create_all(),老数据库里的表不会自动更新,你还得手动删库重建,开发时的数据就全没了。

用了迁移工具后,模型改动只需要三步:

flask db migrate -m "add announcements table" flask db upgrade

这样部署到服务器时,表结构就能无缝升级,这是后期省心的关键。

4. 核心功能实战:从登录到选课完成的完整实现

理论讲完了,我们来把这个系统一步步拼起来。以下所有代码都是我实际跑过的简化版,可以直接抄进你的Flask项目。

4.1 登录与会话管理

登录功能的关键是不要明文存密码。我用werkzeug.security的generate_password_hash来做密码哈希:

from werkzeug.security import generate_password_hash, check_password_hash def create_user(username, password, role=2): user = User( username=username, password_hash=generate_password_hash(password), role=role ) db.session.add(user) db.session.commit()

登录时用check_password_hash验证:

def login(username, password): user = User.query.filter_by(username=username).first() if user and check_password_hash(user.password_hash, password): session['user_id'] = user.id session['username'] = user.username return True return False

这里用的是服务端会话,Flask默认的session是基于客户端Cookie的,但它存的是签名后的数据,不能篡改,所以存一个user_id进去是安全的。更好的方案是用Flask-Login扩展,它会帮我们处理is_authenticated这些状态,我建议直接上手就用Flask-Login,省去自己维护session的麻烦。

4.2 学生选课接口的完整实现

我在第2章已经给了select_course的核心代码,现在补充时间窗口检查的逻辑。假设我们有一个config表或配置文件里存了选课开始和结束时间:

def select_course_api(course_id): if not is_selection_open(): return {'code': 400, 'msg': '不在选课时间窗口内'} result = select_course(course_id, current_user.id) return result

is_selection_open的逻辑不复杂,就是“当前时间 >= start_time 且 <= end_time”。但要注意,如果你在多台机器上部署,时间同步很重要,各服务器时间不一致会导致有的学生能选有的不能选。单机部署一般没这个问题,但值得注意。

4.3 退课接口的并发考虑

退课就是选课的逆操作,但同样要注意并发问题:

def drop_course(course_id, student_id): course = Course.query.filter_by(id=course_id).with_for_update().first() selection = Selection.query.filter_by(course_id=course_id, student_id=student_id).first() if not selection: return {'code': 400, 'msg': '未选此课程'} db.session.delete(selection) course.selected_count = max(0, course.selected_count - 1) db.session.commit() return {'code': 200, 'msg': '退课成功'}

这里加with_for_update()是为了防止这样一种情况:某个学生同时打开两个标签页,一个在退课一个在选课,由于事务隔离级别的原因,可能出现数据混乱。加锁后,这两个操作会串行执行。

4.4 课程列表的分页查询

当课程数量多了以后,前端一次性渲染所有课程会很卡。我给课程列表接口加了分页参数:

@app.route('/api/courses') def course_list_api(): page = request.args.get('page', 1, type=int) per_page = request.args.get('per_page', 10, type=int) pagination = Course.query.order_by(Course.id.asc()).paginate( page=page, per_page=per_page, error_out=False ) courses = [{ 'id': c.id, 'name': c.name, 'teacher': User.query.get(c.teacher_id).username, 'capacity': c.capacity, 'selected_count': c.selected_count, 'schedule': c.schedule, 'left_count': c.capacity - c.selected_count } for c in pagination.items] return jsonify({'total': pagination.total, 'items': courses})

分页不只是为了界面好看,更是为了减少数据库的查询压力。如果你把所有课程一次查出,比如500门课,加上联表查询、序列化,接口响应可能要2秒,用户感受非常差。分页后每次只查10条,响应基本在50ms以内。

4.5 前端页面怎么做

很多人纠结前端要不要用Vue或React。个人建议:课程设计级别的系统,用服务端渲染的Jinja2模板就够了,别引入前后端分离的复杂度。

我在项目中用的是Jinja2模板 + Bootstrap,页面不用写很多JavaScript。核心的动态逻辑就是点击选课按钮后发一个POST请求,然后刷新列表。比如选课按钮的处理:

function selectCourse(courseId) { fetch('/api/select_course', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({course_id: courseId}) }) .then(res => res.json()) .then(data => { if (data.code === 200) { alert('选课成功'); // 可选:刷新页面或更新余量 location.reload(); } else { alert(data.msg); } }); }

这样最简单也最稳定。如果你要追求体验更好,可以学一点Vue的v-if/v-for,但核心教训是:不要把简单系统搞复杂化,能用模板解决的不用框架。

5. 部署与运行:从本地到服务器的完整路径

一个系统写完了,最终要跑起来给别人用。这一节我把从本地开发到服务器部署的完整过程捋一遍。

5.1 本地环境准备

假设你在一台干净的机器上,Python 3.10+:

python -m venv venv source venv/bin/activate # Windows 上用 venv\Scripts\activate pip install flask flask-sqlalchemy flask-migrate flask-login

建议把所有依赖用requirements.txt固化下来:

pip freeze > requirements.txt

这样换环境时只要一个命令:

pip install -r requirements.txt

5.2 用SQLite还是MySQL

开发环境用SQLite就够了,零配置,一个文件搞定。但部署到服务器时,如果预期并发不高(比如校内课程设计),继续用SQLite其实也够。如果选课时间集中且访问量大,建议换MySQL,因为MySQL的SELECT ... FOR UPDATE行锁机制更健壮,SQLite虽然也支持行锁,但在高并发写场景下会有“数据库被锁定”的报错。

换MySQL的改动很小,只要改数据库连接字符串:

SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://user:password@localhost/course_system'

然后生成迁移、建表:

flask db upgrade

5.3 用Gunicorn部署Flask应用

Flask自带的开发服务器app.run()只能用于调试,部署时一定要让它跑在真正的WSGI服务器上,比如Gunicorn。启动命令:

gunicorn -w 4 -b 0.0.0.0:8000 app:app

-w 4表示开4个worker进程,这是生产环境的基本配置。用多worker后,你要注意内存型状态的共享问题,比如基于进程内字典的会话会失效,所以前面才强调要用数据库做会话存储或使用签名Cookie。

5.4 Nginx反向代理

服务器上建议再套一层Nginx,做反向代理和静态文件服务。Nginx配置里最核心的一段:

server { listen 80; server_name course.example.com; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } }

Nginx的作用不只是让请求转发更高效,还能拦截静态文件请求,减轻Python进程的压力。前端用的CSS、JS、图片全都可以丢到Nginx直接返回,Python只处理接口请求。

6. 踩坑记录与方法论:你能遇到的雷我都帮你趟过了

这一节是全文价值最高的部分。我在开发过程中踩了不少坑,整理出来给你参考,可能几句话就能帮你省下半天时间。

6.1 并发超卖问题的排查经历

我第一次写完选课接口后,做了个简单测试:用脚本模拟50个学生同时抢同一门只有20个名额的课程,结果选成功的人数超过了20。当时的代码就是简单先查询再判断,没有加行锁。

排查过程:我先用日志打印出每个人的选课时间,发现大量请求都在同一个毫秒级内读取了selected_count=19,然后各自加1写回,导致最终数据变成29。这就是经典的“读改写”竞态。解决方案也就是前面写的with_for_update(),加锁后我重新测试,20个名额最终只成功20人,问题解决。

这个坑的教训是:只要涉及共享资源的“先查后改”,在并发环境下必须加锁或使用原子操作。

6.2 密码安全问题

我见过太多课程设计项目里,用户表里直接存明文密码。一旦数据库泄露,所有用户的密码就全裸奔了。有同学觉得“我这只是课程设计,没人会攻击”,但这种习惯会让你进入职场后付出代价。从第一行代码开始养成安全习惯,用generate_password_hash做哈希,成本几乎为零,为什么不做呢?

6.3 表单验证的空指针问题

另一个高频Bug是提交表单时没做字段校验,比如选了课程就提交、没填课程名就建课。这个问题在Flask里可以用validate字段或手动判断搞定:

if not course_name or not teacher_id: return jsonify({'code': 400, 'msg': '参数不完整'})

看似是基本功,但很多人因为没加这段代码,导致后面报错时满脸迷惑。

6.4 时间处理时区问题

选课时间窗口我用的是datetime.now(),返回的是运行机器的本地时间。如果服务器用的是UTC时间,而学生在中国时区,就会出现“选课系统明明已经开始,但显示没到时间”的情况。

经验做法是:统一使用datetime.utcnow()存储,前端显示时再转本地时区。或者干脆把服务器时区设置成和用户一致,比如TZ=Asia/Shanghai,一步到位。这块我建议项目一开始就明确规范,不然后期就是无底洞。

6.5 接口返回统一格式

我之前写接口时,有的是{code: 200, data: ...},有的直接返回一个裸列表,导致前端处理起来很痛苦。后来我统一了响应格式:

{ "code": 200, "msg": "success", "data": {} }

所有正常返回code=200,业务错误返回其他业务码,系统异常返回500。前端对code统一判断,只有200才取data。这个规范最好第一天就定下来。

7. 扩展思路:这个系统还能怎么升级

做完基础功能后,如果你的课程设计想拿高分,或者实际业务需要进一步迭代,以下几个方向可以考虑。

7.1 增加选课前志愿排队模式

真实大学的选课系统为了缓解热门课抢不到的问题,会引入“志愿制”——学生先预选志愿,然后系统统一抽签。这种模式在技术上增加一个applications表和一个后台定时任务(用APScheduler或Celery),到了选课截止时间自动处理抽签,把名额随机分配给中签学生。

7.2 缓存与异步化

如果选课系统的访问量到了千级并发,MySQL扛不住频繁的SELECT ... FOR UPDATE,就需要引入Redis做缓存。把课程余量放在Redis里,用DECR命令原子扣减,扣减成功后再异步落库。但异步落库的坑也多,比如掉单问题,需要你仔细设计对账机制。

7.3 监控与日志

加一个简单的操作日志表,记录每次选课、退课、登录等敏感操作,出问题时能回溯。更进一步可以接入Sentry,当系统抛异常时自动上报。我当时给系统加了一个简单的日志中间件:

@app.before_request def log_request(): path = request.path if path.startswith('/api/'): write_operation_log(request)

这样每次API请求都会留下痕迹,排查问题效率高了一截。

7.4 你一定会用到的目录结构

最后给出一份我推荐的项目目录结构,别把代码全堆在一个文件里:

course_system/ ├── app.py # 入口,创建Flask实例 ├── config.py # 配置项 ├── requirements.txt ├── models/ │ ├── __init__.py │ ├── user.py │ ├── course.py │ └── selection.py ├── views/ │ ├── __init__.py │ ├── auth.py │ ├── course_api.py │ └── admin.py ├── utils/ │ ├── __init__.py │ ├── auth_utils.py │ └── response.py └── templates/ ├── base.html ├── login.html ├── course_list.html └── admin_dashboard.html

这样拆,每个文件职责单一,查Bug时不用翻几百行代码。经验之谈:让代码结构清晰,从第一天就要养成习惯。哪怕你现在觉得项目小、没必要,但写到后面功能越来越多的时候,你会感谢自己当初的坚持。

我在实际开发中还发现一个很实用的小技巧:凡是涉及金额、名额、库存这类要精确的字段,一律别用浮点数,用整数或者Decimal类型。Python的浮点数在计算时会有精度损失,比如0.1 + 0.2 != 0.3,这在普通场景没事,但在资源计数场景会出大问题。选课系统的名额我用整数,问题就少很多。

好了,整个在线选课系统的完整实现链路就讲到这儿,希望对你做类似项目有实实在在的帮助。踩过这一遍坑,以后再遇到什么预约、抢券、报名系统,核心逻辑都是相通的,拿这套设计改改就能用。

返回列表