简介这是一份基于Flask与MySQL实现的在线OJ评测平台项目面向Python开发者、Web后端学习者及需要快速搭建判题服务的师生覆盖从题目管理、代码提交到自动评测的完整业务链路。压缩包内含完整项目源码、部署文档与数据资料部署文档详细说明Python环境配置、依赖安装及启动步骤数据资料可直接初始化数据库减少上手成本。资源共278个文件以py源码、html模板、scss/less样式、js脚本为主同时带有pyc/pyd编译文件、dll动态库及readme等辅助说明整体8.49MB目录结构清晰便于定位与二次开发。目前已有125人学习按文档操作即可运行适合作为课程设计、毕业设计或在线判题系统入门参考。1. FlaskMySQL 组合做 OJ 评测平台真正的核心难点在评测链路一个能用的 OJ 评测平台最容易被低估的部分不是 Web 页面而是“评测”本身。用户提交一段 C 语言代码系统要完成编译、运行、限制资源、比对输出、回写数据库这一条链路如果设计不好就会出现提交卡死、判题结果和库里 status 对不上、多个评测进程抢同一道任务等怪问题。标题里 FlaskMySQL 的组合其实把边界切得很清楚Flask 管 HTTP 与业务逻辑MySQL 管题目、用户、提交记录这些数据的持久化评测器独立成一个后台任务去消费提交记录。这样的结构既能让新手按模块读完源码也能让有经验的工程师快速替换掉某个环节而不影响其他部分。本文就按这条链路展开先讲 MySQL 表结构如何为评测服务再写评测核心怎么落地最后聊队列调度和部署参数。2. MySQL 表结构和 Flask 模型先把用户、题目、提交三类数据理清2.1 评测平台的核心表users、problems、submissions、test_cases一个典型 OJ 平台的 MySQL 表不会特别多核心就四张用户表、题目表、提交表、测试点表。另加比赛表、比赛报名表看具体业务是否要求。设计表结构时多想想submissions这张表的读写频率用户每次提交都会插入一行评测进程更新 status排行榜和提交记录又频繁按题目和时间查询。表结构不合理后面加索引也救不回来。下面是建表脚本去掉了外键约束便于高并发写入场景下调优CREATE TABLE users ( id INT UNSIGNED PRIMARY KEY AUTO_INCREMENT, username VARCHAR(64) NOT NULL UNIQUE, password_hash VARCHAR(128) NOT NULL, role TINYINT NOT NULL DEFAULT 0 COMMENT 0普通用户 1管理员, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE problems ( id INT UNSIGNED PRIMARY KEY AUTO_INCREMENT, title VARCHAR(128) NOT NULL, description TEXT NOT NULL, input_desc TEXT, output_desc TEXT, time_limit INT NOT NULL DEFAULT 1000 COMMENT 毫秒, memory_limit INT NOT NULL DEFAULT 65536 COMMENT KB, is_visible TINYINT NOT NULL DEFAULT 1, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE submissions ( id BIGINT UNSIGNED PRIMARY KEY AUTO_INCREMENT, user_id INT UNSIGNED NOT NULL, problem_id INT UNSIGNED NOT NULL, language VARCHAR(16) NOT NULL COMMENT c/cpp/java/python, code MEDIUMBLOB NOT NULL, status TINYINT NOT NULL DEFAULT 0 COMMENT 0排队 1评测 2..7见状态枚举, score SMALLINT NOT NULL DEFAULT 0, run_time INT NOT NULL DEFAULT 0 COMMENT 毫秒, run_memory INT NOT NULL DEFAULT 0 COMMENT KB, error_info TEXT, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, INDEX idx_user_id (user_id), INDEX idx_problem_id (problem_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;代码里有几个点值得说明code字段用MEDIUMBLOB而不是TEXT是为了保留用户提交的原始二进制内容避免字符集转换导致编译失败status用 TINYINT 是因为状态枚举量很小不必要用VARCHAR存字符串created_at用 TIMESTAMP 可以自动带时区处理。题目表和测试点表之间通过problem_id关联测试点数据另存不塞进description里。2.2 用 Flask-SQLAlchemy 在 models.py 里完成 MySQL 映射源码包里常见的做法是单独建一个models.py所有表模型集中放置。Flask-SQLAlchemy 的模型定义如下注意和上面 SQL 字段保持对齐from flask_sqlalchemy import SQLAlchemy from datetime import datetime db SQLAlchemy() class Submission(db.Model): __tablename__ submissions id db.Column(db.BigInteger, primary_keyTrue, autoincrementTrue) user_id db.Column(db.Integer, nullableFalse, indexTrue) problem_id db.Column(db.Integer, nullableFalse, indexTrue) language db.Column(db.String(16), nullableFalse) code db.Column(db.LargeBinary, nullableFalse) status db.Column(db.SmallInteger, nullableFalse, default0, indexTrue) score db.Column(db.SmallInteger, default0) run_time db.Column(db.Integer, default0) run_memory db.Column(db.Integer, default0) error_info db.Column(db.Text) created_at db.Column(db.DateTime, defaultdatetime.now) def to_dict(self): return { id: self.id, user_id: self.user_id, problem_id: self.problem_id, language: self.language, status: self.status, score: self.score, run_time: self.run_time, run_memory: self.run_memory, }这里的to_dict()是给 Flask 接口做 JSON 序列化用的避免在视图函数里反复手动拼字典。status字段加了indexTrue配合后面的组合索引能让排行榜查询避免全表扫描。需要提醒的是db.session的生命周期由 Flask-SQLAlchemy 自动管理但在下面的评测 Worker 里我们会手动commit()因为 Worker 不在请求上下文里运行。2.3 题目与测试点分开存避免评测逻辑被 JSON 字段锁死看过一些课程设计的 OJ 源码喜欢把测试点直接写成 JSON 塞进 problems 表[{input: 1 2, output: 3}, {input: 3 4, output: 7}]这种做法对只有三五个测试点的题目勉强可行一旦题目数量变多、测试点需要按权重计分或者需要单独修改某个测试点而重新评测历史提交JSON 字段的方案就会变得很难维护。评测进程每跑一道题都要解析整个 JSON还要保证 JSON 结构不被非法修改。正规一点的源码包都会单独建test_cases表CREATE TABLE test_cases ( id INT UNSIGNED PRIMARY KEY AUTO_INCREMENT, problem_id INT UNSIGNED NOT NULL, input_data MEDIUMBLOB NOT NULL, output_data MEDIUMBLOB NOT NULL, score_weight INT NOT NULL DEFAULT 100 COMMENT 该测试点分值权重, is_sample TINYINT NOT NULL DEFAULT 0 COMMENT 是否作为样例展示, INDEX idx_problem_id (problem_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;测试点文件内容存数据库省去文件系统同步到多台评测机的麻烦。如果追求更高性能可以把input_data和output_data放到评测机本地磁盘数据库只存文件路径。这两种方案各有取舍源码包常见是前者因为部署简单单机场景完全够用。2.4 提交表索引设计解决 OJ 最常见的慢查询OJ 平台流量一大慢查询会集中在两类一类是按题目查所有提交记录另一类是按用户查历史提交。最典型的语句是SELECT id, user_id, status, run_time, run_memory FROM submissions WHERE problem_id 101 AND status 6 ORDER BY id DESC LIMIT 20;如果只建了idx_problem_idMySQL 会先把该题所有提交捞出来再做status过滤和文件排序。题目提交量上万之后语句就会变慢。实际操作是在submissions上创建组合索引让查询能直接定位到目标区间ALTER TABLE submissions ADD INDEX idx_problem_status_id (problem_id, status, id);这个索引利用 B 树的有序性problem_id和status定位到一组记录后id本身按升序排列MySQL 从后往前扫 20 条就能拿到结果不需要额外 filesort。同理如果经常按用户维度查提交历史可补一个(user_id, problem_id)索引。有些部署文档还建议在created_at上单独建索引但 OJ 很少按时间范围拉全量数据实际收益不大反而增加写入成本。3. 评测核心的编写编译、运行、比对输出评测平台里最难的一层3.1 为什么评测不能放在 Flask 的请求处理线程里源码包如果只把 Flask 和 MySQL 粘在一起而不去管评测发生的位置那一定是个半成品。评测任务是典型的长耗时操作C 代码编译可能要几百毫秒到几秒程序运行可能直到超时才被中断。如果直接在 Flask 路由里同步执行前端发一个提交请求HTTP 连接就会一直挂着直到评测结束才返回。这种设计在面对两个并发用户提交时就会把 Flask 开发服务器线程占满。正确做法是把评测拆成独立模块。Web 进程只负责把提交写入submissions表然后立刻返回“排队中”。后台 Worker 用自己的循环去 MySQL 里领任务执行评测再把结果写回同一行记录。前端通过轮询接口查看 status 变化。这种架构下 Web 进程和评测进程可以单独扩展也是本标题源码包最核心的结构决策。3.2 一个最小可执行的 judge_one 函数评测器最核心的函数可以抽象为“输入源代码路径、语言、测试点数据返回判定结果”。下面这段代码适合在 Linux 环境下直接运行作为理解整个评测链路的起点import subprocess import resource import time import os TIME_LIMIT_MS 1000 MEMORY_LIMIT_KB 65536 def normalize_output(raw: bytes) - str: 去掉行尾空白统一换行符压缩连续空行 text raw.decode(utf-8, errorsignore) lines [line.rstrip() for line in text.splitlines()] while lines and lines[-1] : lines.pop() return \n.join(lines) \n def limit_memory(): 子进程启动时设置虚拟内存上限单位为字节 resource.setrlimit(resource.RLIMIT_AS, (MEMORY_LIMIT_KB * 1024, MEMORY_LIMIT_KB * 1024)) def judge_one(problem_id, language, source_code, input_data, expected_output): workdir /tmp/oj_ str(os.getpid()) _ str(time.time()) os.makedirs(workdir, exist_okTrue) source_path os.path.join(workdir, main.c if language c else Main.java) with open(source_path, w) as f: f.write(source_code) if language c: # 编译参数 -O2 -lm 是常见配置-O2 会稍微拉长编译时间 cp subprocess.run( [gcc, source_path, -o, os.path.join(workdir, main), -O2, -lm], capture_outputTrue, cwdworkdir ) if cp.returncode ! 0: return {status: compile_error, error: cp.stderr.decode(utf-8, errorsreplace)} run_cmd [os.path.join(workdir, main)] else: return {status: unsupported_language} try: start time.time() rp subprocess.run( run_cmd, inputinput_data.encode(), capture_outputTrue, timeoutTIME_LIMIT_MS / 1000, cwdworkdir, preexec_fnlimit_memory, env{PATH: /usr/bin:/bin} ) except subprocess.TimeoutExpired: return {status: time_limit_exceeded} cost_ms int((time.time() - start) * 1000) if rp.returncode ! 0: # 内存超限通常表现为进程被系统杀死stderr 里可能包含 memory 字样 return {status: runtime_error, error: rp.stderr.decode(utf-8, errorsreplace)[:500]} if normalize_output(rp.stdout) normalize_output(expected_output): return {status: accepted, time_ms: cost_ms} return {status: wrong_answer, time_ms: cost_ms}这段代码有几个要点preexec_fnlimit_memory会在子进程执行前调用setrlimit这是 Linux 上限制用户程序内存最常用的方式不需要额外安装 cgroup 工具timeout参数是 Python 层面的超时保护到点直接抛异常并终止子进程env被重置为最小化环境避免用户代码读到宿主机上的环境变量。对评测进程而言subprocess.run的capture_outputTrue会先把输出读进内存所以输出上限也要限制否则恶意程序可以无限打印拖垮评测机。3.3 C、Java、Python 的编译运行参数对照不同语言的评测参数差异很大下面这张表是部署 OJ 时最常用的一组配置语言编译命令运行命令关键注意事项Cgcc main.c -o main -O2 -lm./main默认栈空间较小递归深的题目需ulimit -sCg main.cpp -o main -O2 -lm./main同 C注意 libstdc 版本Javajavac Main.javajava -Xmx256M Main主类必须叫Main-Xmx要小于平台内存限制Python3不编译python3 -B main.py-B禁止生成__pycache__避免污染评测目录Java 的内存限制需要双重设置外层RLIMIT_AS给 JVM 进程设上限内层-Xmx让 JVM 自己在堆内存分配时自我约束。Python 则需要额外防范死循环和内存暴涨timeout对付死循环没问题但 Python 进程在内存吃满时会被系统 OOM Killer 杀掉此时returncode为负值需要单独判断。3.4 输出校验的边界行尾符、空行和答案错误很多新手写的评测对比是if output expected在 Windows 上开发的题目数据传到 Linux 后会发现答案错误因为文件里的\r\n没有被处理。更稳的做法是统一归一化每行去掉尾部空格忽略文件末尾多余空行把\r\n统一成\n。上面normalize_output做的就是这个事。还需要注意特判题目的存在。例如精度题要求误差小于 1e-6字符串题允许大小写不敏感这些不能走标准输出比对逻辑。常见做法是给problems表加一个special_judge字段非空时评测器调用题目指定的特判程序把用户输出和标准答案同时交给特判程序由它返回 0 或 1。4. 评测队列设计与 Flask 的 API 对接让提交不再卡住 Web 进程4.1 用 MySQL 表当任务队列而不是 Python 内存队列评测任务的排队方式有几种选择直接用 Pythonqueue.Queue、上 Redis、或者就靠 MySQL 表本身。内存队列的优点是响应快缺点是 Web 进程一旦重启未评测的任务全部丢失而且多 worker 进程之间各自维护一个队列会产生重复领取的问题。对于标题这种 FlaskMySQL 的技术栈用 MySQL 表本身做队列是最不自找麻烦的方式。submissions表里status0的记录就是待评测任务Worker 每次去取一条status0且id最小的记录取到后立刻把status改成 1。相当于数据库既存业务数据又充当任务队列。不需要引入额外组件部署时少一个故障点。队列方案任务持久化部署复杂度多 Worker 竞争处理Python queue.Queue无低不支持跨进程MySQL 表轮询有低需要 SKIP LOCKEDRedis List有中天然原子弹出4.2 Worker 轮询循环的正确打开方式评测 Worker 是一个独立进程和 Flask 应用分开启动。它不断查询待评测任务然后调用上一章的judge_one。轮询循环要注意两个问题控制轮询频率避免频繁请求 MySQL每个任务结束后清理数据库连接防止连接泄漏。import time from app import create_app, db from models import Submission from judge import judge_one app create_app() def dispatch(sub): 根据语言调用对应评测逻辑完整版需要按 sub.problem_id 拉取测试点 return judge_one(problem_idsub.problem_id, languagesub.language, source_codesub.code.decode(utf-8, errorsignore), input_data, expected_output) def worker_loop(): while True: # Worker 必须进入应用上下文才能用 db.session with app.app_context(): sub (Submission.query .filter_by(status0) .order_by(Submission.id.asc()) .first()) if sub is None: time.sleep(0.5) continue sub.status 1 # 标记为评测中防止其他 Worker 重复领取 db.session.commit() result dispatch(sub) sub.status STATUS_MAP[result[status]] sub.run_time result.get(time_ms, 0) sub.error_info result.get(error, ) db.session.commit() time.sleep(0.05)注意with app.app_context()的位置。Submission.query依赖 Flask-SQLAlchemy 的上下文绑定脱离上下文调用会报Working outside of application context错误。实际项目里还会把测试点数据读取放在dispatch里根据problem_id查出test_cases表逐个执行judge_one并累加 score这里做了简化处理。4.3 状态机约定与前端轮询查询前端页面需要知道 status 每个整数代表什么含义这需要在源码里统一维护一份枚举。下面是 OJ 平台最常见的一套状态约定status含义是否终态0Pending 排队中否1Judging 评测中否2Compile Error 编译错误是3Wrong Answer 答案错误是4Time Limit Exceeded 超时是5Memory Limit Exceeded 超内存是6Accepted 通过是7Runtime Error 运行错误是Flask 提供给前端的查询接口不需要返回整个code字段那个只在用户查看自己源码时才需要。提交记录接口一般返回to_dict()结果前端拿到 status 后根据枚举自行渲染颜色和文案。from flask import jsonify app.route(/api/submission/int:submission_id) def api_submission(submission_id): sub Submission.query.get(submission_id) if sub is None: return jsonify({error: submission not found}), 404 return jsonify(sub.to_dict())这个接口只做一次主键查询MySQL 使用聚簇索引即使submissions表有几百万行也能毫秒级返回。前端通常会设置 1~2 秒的轮询间隔直到 status 进入终态。4.4 多 worker 并发抢同一提交的坑SKIP LOCKED如果部署时用supervisor同时起了两个评测 Worker上面的.first()可能被两个进程同时查出同一条记录两边都去评测浪费算力且可能写乱结果。MySQL 8.0 提供了FOR UPDATE SKIP LOCKED专门解决任务队列场景下的并发抢占SELECT id, user_id, problem_id, language, code FROM submissions WHERE status 0 ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED;SKIP LOCKED会让被其他事务锁住的行直接跳过而不是等待锁释放。这样多个 Worker 同时拉任务时每个人拿到的都是不同的行。如果用 SQLAlchemy 表达是sub (Submission.query .filter_by(status0) .order_by(Submission.id.asc()) .with_for_update(skip_lockedTrue) .first())MySQL 5.7 没有SKIP LOCKED需要在事务里手动模拟先用UPDATE把一条记录的 status 改成 1再查询。同时要处理 Worker 崩溃后 status1 永远卡住的问题。常见的兜底方案是加updated_at启动时把所有status1且updated_at在 10 分钟之前的记录重置回 0这个过程也可以写成 MySQL 存储过程定时执行效果是一样的。5. 部署时让 MySQL 连接池和评测隔离性同时稳住的几个设置5.1 Flask-SQLAlchemy 连接池参数与 gunicorn 多进程的配合用 gunicorn 启动 Flask 时如果-w 4起了 4 个 worker 进程每个进程都会维护一份独立的数据库连接池。默认连接池大小为 5高峰期 4 个进程最多占 20 个连接对 MySQL 本身不构成压力但wait_timeout默认 8 小时会导致连接被服务端断开后Flask 还在用旧连接报MySQL server has gone away。部署时在 Flask 配置里显式设置四个参数SQLALCHEMY_ENGINE_OPTIONS { pool_size: 10, max_overflow: 20, pool_recycle: 3600, pool_pre_ping: True, }pool_recycle让连接在 1 小时后主动重建避开 MySQL 的 8 小时超时pool_pre_ping每次从连接池取连接前发送一个 SELECT 1 探活虽然多了一次网络往返但对评测平台这种低吞吐场景完全可接受。gunicorn 启动命令一般是这样gunicorn -w 4 -b 0.0.0.0:5000 --timeout 60 app:app--timeout 60是给 HTTP 请求设置的并非评测超时。评测 Worker 单独用 supervisor 或 systemd 拉起来不要混在 gunicorn 里。5.2 五分钟验证评测隔离性故意提交一个死循环部署完成后建议立刻做一组验证确认评测超时和内存限制真的生效。先提交一段死循环代码#include stdio.h int main() { while (1) {} return 0; }然后在评测机器上看进程列表ps aux | grep main正常情况下评测进程会在 1 秒左右被timeout杀死数据库里该条 submission 的 status 变成 4run_time接近 1000 而不是继续增长。再用一段不断申请内存的代码验证RLIMIT_AS#include stdlib.h #include string.h int main() { while (1) { char *p malloc(1024 * 1024); memset(p, 0, 1024 * 1024); } }这条记录最终会返回 Runtime Error 或 Memory Limit Exceeded取决于你按退出码判断还是按 stderr 内容判断。两种结果都是可接受的重要的是单个评测任务不会拖垮整个 Worker 进程。5.3 评测进程本身的口子要收小评测时通过preexec_fn设置RLIMIT_AS和RLIMIT_CPU之外还有几个容易被忽略的设置给评测用户创建独立系统账号用preexec_fn切换uid和gid禁止用户代码访问网络方法是在子进程启动前调用setsid配合 iptables 规则。单机部署阶段至少要做到禁止用户代码通过子进程再拉新进程最简单的办法是把进程数上限RLIMIT_NPROC一并设掉def limit_proc(): resource.setrlimit(resource.RLIMIT_NPROC, (10, 10))这几个resource设置组合起来能让整个评测环境在单机场景下达到可接受的隔离程度再往上走就是 Docker 容器评级但那就超出标题里 FlaskMySQL 的讨论范围了。部署文档里如果只贴了 gunicorn 和 MySQL 建表语句说明还没有意识到评测隔离的重要性动手前一定要补上这一层。本文还有配套的精品资源点击获取