用Python构建自习室座位预约系统:状态机+事务+定时任务

发布时间:2026/10/9 6:18:31
用Python构建自习室座位预约系统:状态机+事务+定时任务
自习室座位预约这个需求很多人可能第一反应觉得“不就是做个选座界面吗”但真正把系统跑起来你会发现难点全在细节里座位状态怎么保持一致、占座不来的座位由谁释放、高峰期一堆人同时抢同一个位置该怎么处理。我用 Python 从零实现了一套开放自习室座位预约管理系统从需求梳理到数据库设计再到核心逻辑落地把整个过程和踩过的坑都整理在这篇里准备做类似毕业设计、课程设计或者想给小型自习室搭一套预约系统的朋友可以直接照着这个思路走。这个项目表面上是一个 Web 管理系统本质上是一套“状态机 事务控制 定时任务”的组合。座位不是简简单单一个“空/不空”的字段它会在空闲、预约锁定、使用中、暂离、禁用这些状态之间来回切换每一次切换都要保证并发场景下不冲突。我会从最前端的业务规则开始讲然后是技术选型的取舍再到数据库设计和核心代码实现最后把实际运行中遇到的典型案例整理成排查指南。1. 先想清楚要解决什么问题——需求与方案设计1.1 开放自习室的真实痛点我调研过几家高校图书馆和社区自习室发现“开放自习室”的预约难度比想象中大。核心痛点有三个一是占座成本极低。很多人把书往桌上一放就去吃饭、去办事一占就是两三个小时真正想学习的人找不到位置管理员又不知道哪些座位是“有人但不在”。二是高峰期供需错配。考研季、期末周座位利用率能到 95% 以上但同一楼层的不同区域冷热差异很大靠人工引导根本不现实需要系统实时展示哪些区域有位置。三是预约行为需要约束。如果没有违约机制预约了不来、超时不来的人会把座位白白锁死。系统必须能记录每一次预约的履约情况把故意占着不放的行为识别出来。所以这套系统的核心目标不是“做出一个选座页面”而是要做出一套能覆盖“预约—签到—使用—暂离—离座—违约处理”全流程的规则引擎。1.2 角色与核心业务流程设计系统按角色分两类用户业务流完全围绕座位状态展开。学生端主流程是这样的注册登录后按区域、日期查询可预约座位选择空闲座位提交预约申请系统锁定该座位生成预约记录到馆后在终端或手机上签到座位状态变为“使用中”学习过程中需要临时离开可点击“暂离”座位保留一段时间学习结束离座释放座位预约记录归档管理端流程相对简单但非常重要维护座位基础信息位置、区域、是否可用查看所有座位当前状态和历史预约记录处理异常情况如座位设备故障直接禁用该座位统计违约数据必要时对用户进行限制预约处罚整套流程设计有一个原则系统里的每个操作都要影响至少一个状态而每个状态变化都要有明确的操作人和时间记录这样才能在出问题时回溯。1.3 容易被忽视但必须定下来的业务规则需求评审时最容易漏掉的就是规则细节。我实际设计时定了一套默认规则你可以按自己场景调整。预约时间段按 30 分钟为一个粒度单次预约最长 4 小时。这个长度的选择参考了图书馆的实际调研——绝大多数人的连续学习时间在 2 到 3 小时之间超过 4 小时的学习者往往需要中途吃饭休息正好对应“暂离”设计。签到时间窗口是预约开始时间前后 30 分钟。注意是“前后”而不是“后”提前到了也可以签到系统提前开始计时但不提前结束。暂离保留时间设 30 分钟超过后座位自动释放预约记录标记为“使用超时释放”这种记录会影响信用分。违约判定采用“30 天内违约满 3 次暂停预约资格 7 天”的规则。违约的定义包括预约后超时未签到、暂离超时未归、恶意反复预约后取消。这些规则看着简单但它直接决定了数据库表结构怎么设计、定时任务跑哪些逻辑。建议开发前先用 Excel 或纸笔画一张状态流转表把所有可能路径列出来再开始写代码。2. 技术选型我为什么用 Python Flask SQLAlchemy SQLite2.1 Web 框架怎么选Python 生态里做 Web 后端主流的就三个Flask、Django、FastAPI。我把它们摆在一起对比过框架核心特点上手难度适合场景Flask轻量灵活扩展自己装低中小型项目、课程设计、快速原型Django全家桶自带 Admin 和 ORM中大型系统、内容管理类、需要后台管理FastAPI异步高性能自动生成 API 文档中前后端分离、高并发接口服务我最后选了 Flask理由很实在。Django 的功能虽然全但对这个项目来说太重了尤其是它的 Admin 后台虽然是亮点但自定义座位状态流转反而被框架的“默认习惯”束缚。FastAPI 的异步模型确实漂亮但自习室预约这种场景的并发量远没到需要异步 IO 来扛的程度用同步逻辑反而更容易理解和调试。Flask 的灵活度正好够用。需要数据库操作就装 SQLAlchemy需要定时任务就加 APScheduler需要表单验证就接 WTForms每个组件都是独立的你可以完全掌控项目结构。2.2 数据存储SQLite 还是 MySQL存储方案是我在开发中后期才认真想清楚的。一开始我也纠结要不要直接上 MySQL后来分析了一下真实需求开放自习室的规模通常是一个校区、一栋楼活跃用户撑死几千人高峰期的并发请求也就几十上百 QPSSQLite 完全扛得住。SQLite 最大的优势是零配置、单文件、备份简单整个数据库就是一个自习室.db 文件拷贝走就能迁移。Python 标准库自带驱动不需要额外装服务。对课程设计和中小型场景这能省很多部署上的事。但 SQLite 有一个明显短板写入锁是数据库级别的多个客户端同时写会报 database is locked。解决方法是开启 WAL 模式让读操作和写操作不互相阻塞再把写事务尽量做得短小。如果你以后要部署到云服务器上、面向几千人同时在线再把 SQLAlchemy 的连接字符串从 sqlite:/// 换成 mysql:// 就行业务代码基本不用动。我在项目里就把数据库访问层单独封装了一层为的就是将来切换数据库时路由和业务逻辑不用改。2.3 项目目录结构好的项目结构能让后面加功能、修 bug 都舒服很多。我的目录是这样的reservation_system/ ├── run.py # 启动入口 ├── config.py # 配置文件 ├── requirements.txt # 依赖清单 ├── app/ │ ├── __init__.py # 应用工厂初始化 Flask 和扩展 │ ├── models.py # SQLAlchemy 数据模型 │ ├── extensions.py # db、scheduler 等扩展实例 │ ├── api/ │ │ ├── __init__.py │ │ ├── auth.py # 注册、登录、Token 校验 │ │ ├── seat.py # 座位查询、预约、签到、释放接口 │ │ └── admin.py # 管理端接口 │ ├── tasks.py # 定时任务 │ ├── utils/ │ │ ├── decorators.py # 登录校验、权限校验装饰器 │ │ └── response.py # 统一响应格式 │ └── templates/ # 前端页面Jinja2 │ ├── index.html │ ├── admin.html │ └── ... ├── tests/ │ ├── test_seat.py │ └── test_reservation.py └── README.md这个结构把“数据模型”“业务接口”“定时任务”“页面展示”分开互不纠缠。尤其是 models.py 独立出来后面改表结构时不会牵连到视图函数。3. 数据库设计与状态流转这才是系统的真正核心3.1 三张核心表怎么设计整个系统的数据模型可以收敛为三张表用户表、座位表、预约记录表。不要一开始就设计一大堆关联表先把核心跑通后续再按需加。用户表简化后是这样的class User(db.Model): __tablename__ user id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(64), uniqueTrue, nullableFalse) password_hash db.Column(db.String(128), nullableFalse) role db.Column(db.String(20), defaultstudent) # student / admin credit_score db.Column(db.Integer, default100) # 信用分 created_at db.Column(db.DateTime, defaultdatetime.now)座位表是整个系统的核心实体class Seat(db.Model): __tablename__ seat id db.Column(db.Integer, primary_keyTrue) seat_no db.Column(db.String(20), uniqueTrue, nullableFalse) area db.Column(db.String(50), nullableFalse) # 区域如 A区一楼 floor db.Column(db.String(20), nullableFalse) status db.Column(db.String(20), defaultidle) # idle 空闲 / locked 预约锁定 / occupied 使用中 / away 暂离 / disabled 禁用 updated_at db.Column(db.DateTime, defaultdatetime.now, onupdatedatetime.now)预约记录表负责承载每一次会话class Reservation(db.Model): __tablename__ reservation id db.Column(db.Integer, primary_keyTrue) user_id db.Column(db.Integer, db.ForeignKey(user.id)) seat_id db.Column(db.Integer, db.ForeignKey(seat.id)) date db.Column(db.Date, nullableFalse) start_time db.Column(db.DateTime, nullableFalse) end_time db.Column(db.DateTime, nullableFalse) status db.Column(db.String(20), defaultpending) # pending 待签到 / checked_in 已签到 / completed 已完成 / cancelled 已取消 / no_show 超时未到 created_at db.Column(db.DateTime, defaultdatetime.now) checkin_time db.Column(db.DateTime) # 实际签到时间 release_time db.Column(db.DateTime) # 实际释放时间注意预约记录表里没有冗余存“座位当前状态”因为座位状态是一个实时变量而预约记录是历史归档两者职责不同。查询某个座位当前是否可用应该直接查 seat.status而不是从预约记录里倒推。3.2 座位状态机的完整流转座位的五态是整个系统最核心的模型。我用一个真实场景把状态流转讲透假设用户 A 在手机上看到座位 S-012 是“空闲”状态提交预约后系统立即把它改为“已锁定”同时生成一条状态为“待签到”的预约记录。此时用户 B 再点这个座位系统会直接提示不可选。A 到自习室后扫码签到座位状态从“已锁定”变为“使用中”预约记录变为“已签到”。A 中途去吃饭在手机上点“暂离”座位状态变为“暂离”系统启动一个 30 分钟倒计时。若 A 在倒计时内回来并点击“归来”座位状态恢复“使用中”若超时未归系统自动把座位恢复为“空闲”预约记录标记为“超时释放”。A 学习结束离开点击“释放座位”座位状态从“使用中”恢复为“空闲”预约记录标记为“已完成”。这里最容易被新手忽略的是座位状态不应该直接从“使用中”跳到“空闲”。中间必须经过一个释放操作或者超时触发否则座位就永远无法被释放其他人只能干等。我在最初版本就踩过这个坑直接改数据库字段把座位恢复了导致好几个预约记录的状态对不上。3.3 预约记录的完整生命周期预约记录的状态定义决定了系统能不能统计“违约率”。我把状态定义如下状态含义进入条件pending待签到用户提交预约成功后checked_in已签到用户预约时间段内签到成功completed已完成用户正常释放座位或预约时间自然结束cancelled已取消用户在签到前主动取消预约no_show爽约签到时间截止仍未签到关键规则是一个座位在同一个时间段内只能存在一条 pending 或 checked_in 状态的预约记录。这个约束不只是写在业务逻辑里还要在数据库层面做唯一性兜底。SQLAlchemy 里可以这样表达from sqlalchemy import UniqueConstraint, func class Reservation(db.Model): __table_args__ ( UniqueConstraint(seat_id, start_time, status, nameuq_seat_time_status), )不过要注意直接把 status 拼进唯一约束会有点问题因为预约取消后状态会变历史记录会冲突。实操中我更推荐用一个“有效标记字段”配合唯一索引__table_args__ ( db.Index(idx_seat_active, seat_id, active_flag, uniqueTrue), )active_flag 这个字段为 1 时表示这条预约记录“正在占用座位”为 0 时表示已归档。这个设计能避免复杂的组合唯一约束查询也更快。4. 核心功能实现与踩坑记录4.1 查询可用座位与预约创建可用座位查询是最高频的接口逻辑本身不复杂但要考虑条件组合。用户可能按区域筛选、按日期筛选还可能需要看到某个座位今天已经被预约了哪些时段。基础的查询可以这样写app.route(/api/seats) def get_seats(): area request.args.get(area) date request.args.get(date, date.today().isoformat()) query Seat.query.filter_by(areaarea) if area else Seat.query # 过滤掉禁用座位 query query.filter(Seat.status ! disabled) seats query.all() return jsonify([seat.to_dict() for seat in seats])这个接口能跑但性能一般因为它没有处理“只想看某时间段有空位的座位”这个场景。高档一点的实现需要反查预约表把该时间段已被预约的 seat_id 排除掉然后返回可预约列表。预约创建是整个系统最需要小心的接口。因为这里会产生并发写操作。4.2 事务与行锁防止大家抢到同一个座位想象一下两个用户同时提交预约都查了一下座位状态是“空闲”然后同时执行更新。如果不加控制最终结果就是两个人预约成功同一个座位。解决方案是在事务内锁定要操作的座位行然后重新读取状态再决定是否更新。这在 SQLAlchemy 里用 with_for_update 实现app.route(/api/reserve, methods[POST]) def reserve_seat(): data request.get_json() seat_id data.get(seat_id) user_id current_user.id # 开启事务 with db.session.begin(): # 锁定该座位行其他事务必须等待 seat db.session.execute( db.select(Seat) .where(Seat.id seat_id) .with_for_update() ).scalar_one_or_none() if seat is None or seat.status ! idle: return jsonify({code: 400, msg: 座位不可预约}) # 再次检查该用户是否已有未完成的预约 active Reservation.query.filter_by( user_iduser_id, active_flag1 ).first() if active: return jsonify({code: 400, msg: 你已有进行中的预约}) # 锁定座位 seat.status locked # 创建预约记录 reservation Reservation( user_iduser_id, seat_idseat_id, start_timedatetime.now(), end_timedatetime.now() timedelta(hours4), statuspending, active_flag1 ) db.session.add(reservation) return jsonify({code: 0, msg: 预约成功})关键点在于 with_for_update 不是简单地把查询结果缓存起来而是告诉数据库“在我提交事务之前其他事务不能读或写这一行”。这样即使两个用户同时进来后一个也会在事务层面被阻塞等第一个提交后再读到最新状态。这个思路我在模拟并发测试中验证过开 20 个线程抢同一个座位最终只有一个成功其余全部返回“座位不可预约”。4.3 定时任务超时未签到与暂离回收座位被锁定后如果用户没来签到就需要定时任务来兜底。我用 APScheduler 实现配置文件里声明任务并启动from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger def release_expired_locks(): 超过签到时间窗口仍未签到的预约自动释放座位 cutoff datetime.now() - timedelta(minutes30) expired Reservation.query.filter( Reservation.status pending, Reservation.start_time cutoff, Reservation.active_flag 1 ).all() for r in expired: seat Seat.query.get(r.seat_id) if seat and seat.status locked: seat.status idle r.status no_show r.active_flag 0 user User.query.get(r.user_id) if user: user.credit_score max(0, user.credit_score - 10) db.session.commit() def release_away_timeout(): 暂离超过 30 分钟的座位自动释放 cutoff datetime.now() - timedelta(minutes30) away_seats Seat.query.filter( Seat.status away, Seat.updated_at cutoff ).all() for seat in away_seats: seat.status idle reservation Reservation.query.filter_by( seat_idseat.id, active_flag1 ).first() if reservation: reservation.status completed reservation.active_flag 0 reservation.release_time datetime.now() db.session.commit() scheduler BackgroundScheduler() scheduler.add_job(release_expired_locks, IntervalTrigger(minutes1), idrelease_expired) scheduler.add_job(release_away_timeout, IntervalTrigger(minutes1), idrelease_away) scheduler.start()这里有一个重要的经验定时任务里要避免复杂操作只做简单的批量更新。最初版本我在任务里调用了发送通知的接口结果通知服务故障整个任务直接中断后面所有释放操作全部堆积。后来我把任务拆成两个独立函数分别处理两类超时即使其中一个挂了另一个也能继续跑。另一个容易踩的坑是时区问题。Python 的 datetime.now() 依赖运行环境时区而数据库里存的可能是 UTC如果前后端时区不一致判断“超时”就会出现半小时甚至一整天的偏差。统一的做法是全部用“同一时区的本地时间”存库任务里和查询里都用 datetime.now()保持口径一致。4.4 暂离机制的实现细节暂离这个功能在设计时我犹豫了很久因为它的业务复杂度比想象中高。后来参考了高校图书馆的通用做法确定了一个原则暂离只对“已签到”的座位开放且同一预约只能暂离一次。接口实现app.route(/api/seat/away, methods[POST]) def seat_away(): 用户点击暂离 reservation get_active_reservation(current_user.id) if not reservation or reservation.status ! checked_in: return jsonify({code: 400, msg: 当前无使用中的预约}) seat Seat.query.get(reservation.seat_id) if seat.status ! occupied: return jsonify({code: 400, msg: 座位状态异常}) seat.status away seat.updated_at datetime.now() db.session.commit() return jsonify({code: 0, msg: 已暂离请 30 分钟内返回})这里注意一个细节为什么要限制“同一预约只能暂离一次”因为如果允许反复暂离就等于变相延长了预约时间用户可以无限续杯。限制一次后暂离就是一次性的“临时离开”行为不会让规则被滥用。“归来”操作是对称的app.route(/api/seat/back, methods[POST]) def seat_back(): reservation get_active_reservation(current_user.id) if not reservation or reservation.status ! checked_in: return jsonify({code: 400, msg: 当前无使用中的预约}) seat Seat.query.get(reservation.seat_id) if seat.status ! away: return jsonify({code: 400, msg: 座位不在暂离状态}) seat.status occupied db.session.commit() return jsonify({code: 0, msg: 欢迎回来})如果用户暂离后直接离开没回来就靠上一节那个 release_away_timeout 任务回收。这里还有一个坑如果暂离超时seat.status 被改回 idle之前的 reservation 也被标记为 completed那你以为的“用户回来点归来”操作会失败因为 get_active_reservation 已经查不到记录。这是正确行为但用户端要有明确的提示“所在座位已被释放请重新预约”。前端交互文案一定要做好不然用户会以为系统出 bug 了。5. 常见问题排查与优化实录5.1 并发抢座SQLite 的锁问题SQLite 在默认回滚日志模式下读操作不阻塞写操作但写操作需要独占数据库并发写会直接报 database is locked。在我用多线程模拟 20 个用户同时抢座时这个问题立刻暴露了。解决办法有两个步骤配合使用。第一步是开启 WAL 模式。在数据库连接后执行from sqlalchemy import event from sqlalchemy.engine import Engine event.listens_for(Engine, connect) def set_sqlite_pragma(dbapi_connection, connection_record): cursor dbapi_connection.cursor() cursor.execute(PRAGMA journal_modeWAL) cursor.execute(PRAGMA busy_timeout5000) cursor.close()WAL 模式允许读操作和写操作并行执行busy_timeout 设置 5 秒当遇到写锁时SQLite 会等待而不是立刻报错。第二步是让事务尽量短小。事务里有网络请求、耗时的复杂查询都会让持锁时间变长进而触发锁冲突。我将预约创建、签到、释放这三个核心操作都控制在十几个毫秒内完成两三个并发事务完全不会有问题。如果上了 MySQLwith_for_update 依然有效但要注意 InnoDB 的行锁是基于索引的如果 where 条件没有走索引行锁会退化成表锁。seat_id 在预约表里一定要建索引。5.2 定时任务不执行或重复执行APScheduler 最容易踩的坑是任务注册在多个进程里。如果你用 uWSGI 或 Gunicorn 启动了多个 worker每个 worker 都会启动自己的 BackgroundScheduler结果是同一个释放任务被多个进程重复执行。虽然我上面的释放逻辑是幂等的查一次状态再改一次但重复执行会浪费资源而且万一中间有个非幂等操作数据就乱了。解决方法是把定时任务独立成一个单独进程运行或者用文件锁确保只有一个实例在执行任务。实际部署时我直接跑了一个 scheduler.py 进程不放在 Web 服务里这样无论 Web 服务怎么扩容任务都只有一份。另一个情况是任务不触发。原因是 BackgroundScheduler 必须在 Flask 应用上下文之外创建但任务函数里又用到了 db.session此时会报“Working outside of application context”。解决办法是在任务函数内部手动创建应用上下文def release_expired_locks(): with app.app_context(): # 业务逻辑 db.session.commit()如果不加 app_context任务虽然被调用了但一执行数据库操作就会抛异常导致任务“看起来没跑”。5.3 前端轮询与跨浏览器兼容前端我用的方案是 Flask 的 Jinja2 模板加原生 JavaScript没有引入前端框架。原因很简单后端重心在预约逻辑上前端不需要单页应用的复杂度。但“座位状态实时更新”这个需求要怎么满足我用了 10 秒轮询每 10 秒请求一次 /api/seats局部刷新座位列表。跨浏览器兼容方面有几个老浏览器容易出问题的点不要用 ES6 的箭头函数写在老浏览器默认脚本里IE11 不支持用 function 语法或者用 Babel 转译fetch 在部分旧浏览器不可用考虑用 XMLHttpRequestCSS flex 和 grid 在 IE 下的兼容问题最好设置 fallback 布局实操中我没有特意追求“全浏览器完美兼容”而是明确了目标浏览器为 Chrome、Edge、Firefox 当前版本在管理后台标注了提示。如果交付给学校用通常现代浏览器已经覆盖 99% 的场景。性能方面如果座位数上千10 秒轮询全量数据也还好一个 JSON 几 KB不算压力。但如果要支持几百人同时在线看座位状态建议改成“轮询接口只返回变化的座位 ID”或者部署一个小型 WebSocket 服务。后者对这个项目来说属于过度设计前期我建议轮询够用就行。5.4 信用分与黑名单机制让规则真正落地信用分功能是我在后来的迭代中加入的。最初的版本只有预约和释放结果发现很多用户预约后不签到也不取消座位数量充足的时候还没什么一到高峰期二十个座位里有六七个是“死锁”的。加入信用分机制后预约行为明显规范了。实现上不复杂在用户表里加一个 credit_score 字段每次违约扣分每周恢复一部分。当信用分低于 60 时预约接口直接拒绝。这里有一个经验处罚规则一定要在用户端“可见”。如果用户不知道自己被扣分、为什么被扣分他只会觉得系统“莫名其妙不让约了”。我在个人中心加了一个“信用记录”模块每次扣分都写一条记录注明原因和时间。上线之后用户主动取消预约的比例明显提高因为大家知道行为会被记录。5.5 座位热力统计数据比想象中更有用系统跑了一段时间后我发现预约记录本身就是一份很有价值的运维数据。我在管理后台加了三个统计报表各区域高峰时段预约量用柱状图展示各座位利用率排名找出常年空置的“冷座位”和天天排队的“热座位”每周违约趋势评估规则是否需要调整比如说如果某个区域的座位利用率长期只有 20%管理员就应该考虑把该区域的部分座位改成灵活座位或者调整开放时间如果某个座位的违约率特别高多半是位置不好或者设备有问题需要去现场查看。这部分功能我用的都是简单的 SQL 聚合查询加 ECharts 图表数据量不大查询很快但对于管理决策很有帮助。这套系统的价值也因此从“预约工具”提升为“管理工具”。最后的几点开发体会这个项目做下来我最深的感受是座位预约系统最复杂的部分不是界面也不是增删改查而是对业务规则的建模和对边界情况的理解。比如“用户预约后主动取消”和“用户预约后超时未到”在业务上性质完全不同一个不算违约一个直接影响信用分但数据库里如果不做区分统计就会失真。另一个体会是宁可在前期花时间把状态机画清楚也不要边写边改。我的第一版代码里座位状态和预约状态是同一个字段后面发现根本没法灵活处理“暂离超时释放”和“正常释放”这两种不同场景被迫回头重构了数据模型。如果你正在做类似项目我建议一上来就把 seat.status 和 reservation.status 分成两个维度各管各的。最后分享一个小技巧把系统跑起来后用自动化脚本模拟 5 天、每天 1000 人的预约行为看看会不会出现“座位被占但预约记录无法对应”的情况。这种并发压力测试能帮你提前找出绝大多数事务逻辑问题比上线后被人发现要省心得多。我后来把压力测试脚本也保留了下来每次改完核心逻辑都会跑一遍到现在已经成了项目的“回归测试”工具。