企业官网建设流程全解析

简介：这是一份基于Flask与MySQL实现的在线OJ评测平台项目，面向Python开发者、Web后端学习者及需要快速搭建判题服务的师生，覆盖从题目管理、代码提交到自动评测的完整业务链路。压缩包内含完整项目源码、部署文档与数据资料，部署文档详细说明Python环境配置、依赖安装及启动步骤，数据资料可直接初始化数据库，减少上手成本。资源共278个文件，以py源码、html模板、scss/less样式、js脚本为主，同时带有pyc/pyd编译文件、dll动态库及readme等辅助说明，整体8.49MB，目录结构清晰，便于定位与二次开发。目前已有125人学习，按文档操作即可运行，适合作为课程设计、毕业设计或在线判题系统入门参考。

1. Flask+MySQL 组合做 OJ 评测平台，真正的核心难点在评测链路

一个能用的 OJ 评测平台，最容易被低估的部分不是 Web 页面，而是“评测”本身。用户提交一段 C 语言代码，系统要完成编译、运行、限制资源、比对输出、回写数据库，这一条链路如果设计不好，就会出现提交卡死、判题结果和库里 status 对不上、多个评测进程抢同一道任务等怪问题。标题里 Flask+MySQL 的组合，其实把边界切得很清楚：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 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; 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 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; 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) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

代码里有几个点值得说明：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_key=True, autoincrement=True) user_id = db.Column(db.Integer, nullable=False, index=True) problem_id = db.Column(db.Integer, nullable=False, index=True) language = db.Column(db.String(16), nullable=False) code = db.Column(db.LargeBinary, nullable=False) status = db.Column(db.SmallInteger, nullable=False, default=0, index=True) score = db.Column(db.SmallInteger, default=0) run_time = db.Column(db.Integer, default=0) run_memory = db.Column(db.Integer, default=0) error_info = db.Column(db.Text) created_at = db.Column(db.DateTime, default=datetime.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字段加了index=True，配合后面的组合索引，能让排行榜查询避免全表扫描。需要提醒的是，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) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

测试点文件内容存数据库，省去文件系统同步到多台评测机的麻烦。如果追求更高性能，可以把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_id，MySQL 会先把该题所有提交捞出来，再做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', errors='ignore') 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_ok=True) 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_output=True, cwd=workdir ) if cp.returncode != 0: return {'status': 'compile_error', 'error': cp.stderr.decode('utf-8', errors='replace')} run_cmd = [os.path.join(workdir, 'main')] else: return {'status': 'unsupported_language'} try: start = time.time() rp = subprocess.run( run_cmd, input=input_data.encode(), capture_output=True, timeout=TIME_LIMIT_MS / 1000, cwd=workdir, preexec_fn=limit_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', errors='replace')[: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_fn=limit_memory会在子进程执行前调用setrlimit，这是 Linux 上限制用户程序内存最常用的方式，不需要额外安装 cgroup 工具；timeout参数是 Python 层面的超时保护，到点直接抛异常并终止子进程；env被重置为最小化环境，避免用户代码读到宿主机上的环境变量。对评测进程而言，subprocess.run的capture_output=True会先把输出读进内存，所以输出上限也要限制，否则恶意程序可以无限打印拖垮评测机。

3.3 C、Java、Python 的编译运行参数对照

不同语言的评测参数差异很大，下面这张表是部署 OJ 时最常用的一组配置：

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 进程之间各自维护一个队列，会产生重复领取的问题。

对于标题这种 Flask+MySQL 的技术栈，用 MySQL 表本身做队列是最不自找麻烦的方式。submissions表里status=0的记录就是待评测任务，Worker 每次去取一条status=0且id最小的记录，取到后立刻把status改成 1。相当于数据库既存业务数据，又充当任务队列。不需要引入额外组件，部署时少一个故障点。

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_id=sub.problem_id, language=sub.language, source_code=sub.code.decode('utf-8', errors='ignore'), input_data='', expected_output='') def worker_loop(): while True: # Worker 必须进入应用上下文才能用 db.session with app.app_context(): sub = (Submission.query .filter_by(status=0) .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 平台最常见的一套状态约定：

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(status=0) .order_by(Submission.id.asc()) .with_for_update(skip_locked=True) .first())

MySQL 5.7 没有SKIP LOCKED，需要在事务里手动模拟：先用UPDATE把一条记录的 status 改成 1，再查询。同时要处理 Worker 崩溃后 status=1 永远卡住的问题。常见的兜底方案是加updated_at，启动时把所有status=1且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 变成 4，run_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 容器评级，但那就超出标题里 Flask+MySQL 的讨论范围了。部署文档里如果只贴了 gunicorn 和 MySQL 建表语句，说明还没有意识到评测隔离的重要性，动手前一定要补上这一层。

本文还有配套的精品资源，点击获取