刷题系统的高并发优化千人同时提交判题队列怎么不炸一、1000 人同时点提交判题系统会发生什么在线判题系统Online Judge有两个流量特征提交是脉冲式的比赛开始时所有人同时提交判题是长耗时的一次判题可能需要数百毫秒到数秒。当 1000 人在比赛开始后同时提交系统瞬间收到 1000 个判题请求——如果不做优化结果不是慢一点而是可能整个系统不可用。判题系统的高并发优化不是简单的加机器而是从架构层面进行多层解耦将接收提交和执行判题分离用消息队列削峰填谷用多级缓存减少重复判题以及用智能调度策略分配评测资源。flowchart TB A[1000 用户同时提交] -- B[API 网关: 限流 鉴权] B -- C[提交服务: 接收并持久化] C -- D[消息队列: 削峰填谷] D -- D1[正常提交队列] D -- D2[优先提交队列: 比赛提交] D1 -- E[判题调度器] D2 -- E E -- F{提交去重?} F --|已缓存| G[直接返回缓存结果] F --|新提交| H[分配评测节点] H -- I[评测节点池] I -- I1[Python 节点1] I -- I2[Java 节点2] I -- I3[C 节点3] I1 -- J[结果回写 通知用户] I2 -- J I3 -- J G -- J二、判题系统的三大性能瓶颈与优化策略瓶颈一评测本身的 CPU 和内存消耗。编译用户代码、在沙箱中执行、比对输出——这些操作都是资源密集的。优化方向是评测节点的水平扩展按语言分池、评测结果的多级缓存相同代码直接返回、以及合理的超时策略死循环不能拖垮节点。瓶颈二消息队列的消费速度。如果评测节点的处理速度跟不上提交速度消息队列会持续堆积。关键的优化手段是根据队列深度动态扩缩评测节点、对等待超时久的任务做降级处理、以及限制单用户的最大并发提交数。瓶颈三数据库写入压力。每次判题结果都需要写入数据库。高并发下写入可能成为新的瓶颈。解决方案是批量写入、异步写入先返回结果给用户再异步持久化、以及使用消息队列缓冲写入请求。三、判题调度器的核心优化实现 高并发判题调度器 核心优化 1. 消息队列削峰——提交与判题异步解耦 2. 多级缓存——相同代码不重复判题 3. 动态扩缩容——根据队列深度调整 worker 数 4. 优先级调度——比赛提交优先处理 import asyncio import time import hashlib import json from dataclasses import dataclass from typing import Dict, Optional from collections import defaultdict from concurrent.futures import ThreadPoolExecutor import aioredis dataclass class JudgeTask: task_id: str problem_id: int language: str source_code: str priority: int 0 # 越大越优先 class HighConcurrencyJudgeScheduler: 高并发判题调度器 为什么用异步架构判题涉及大量 I/O 等待 编译、执行、结果比对异步可以减少线程开销。 def __init__(self, redis_url: str, max_workers: int 8): self.redis None # 延迟初始化 self.redis_url redis_url self.max_workers max_workers self.executor ThreadPoolExecutor(max_workersmax_workers) # 统计指标 self.metrics { submitted: 0, judged: 0, cache_hits: 0, avg_queue_wait_ms: 0.0 } # 去重缓存——TTL 5 分钟 # key hash(source problem_id) self.dedup_cache: Dict[str, str] {} # 动态扩容阈值 self.scale_up_threshold 100 # 队列 100 时扩容 self.scale_down_threshold 10 # 队列 10 时缩容 async def start(self): 启动调度器 self.redis await aioredis.from_url(self.redis_url) # 启动队列消费者 asyncio.create_task(self._consume_loop()) # 启动动态扩缩容监控 asyncio.create_task(self._auto_scale_monitor()) async def submit(self, task: JudgeTask) - dict: 接收提交请求 为什么先检查去重缓存比赛中用户可能 重复提交相同的代码比如连点两次提交按钮 去重避免了无意义的重复判题。 self.metrics[submitted] 1 # 去重检查 dedup_key self._dedup_key(task) if dedup_key in self.dedup_cache: self.metrics[cache_hits] 1 return {status: duplicate, result: self.dedup_cache[dedup_key]} # 写入消息队列带优先级 queue_name judge:high if task.priority 0 else judge:normal await self.redis.lpush( queue_name, json.dumps({ task_id: task.task_id, problem_id: task.problem_id, language: task.language, source_code: task.source_code, submitted_at: time.time() }) ) return {status: queued, task_id: task.task_id} async def _consume_loop(self): 消费判题队列 为什么优先消费高优先级队列比赛场景下 提交延迟直接影响用户体验和公平性。 while True: # 优先消费高优先级队列 task_data await self.redis.brpop( judge:high, timeout0.1) if task_data is None: # 高优先级队列为空消费普通队列 task_data await self.redis.brpop( judge:normal, timeout1) if task_data is None: await asyncio.sleep(0.01) continue task json.loads(task_data[1]) queue_wait_ms (time.time() - task[submitted_at]) * 1000 # 更新统计 self.metrics[avg_queue_wait_ms] ( self.metrics[avg_queue_wait_ms] * 0.9 queue_wait_ms * 0.1 # 指数移动平均 ) # 在独立线程中执行判题避免阻塞事件循环 loop asyncio.get_event_loop() result await loop.run_in_executor( self.executor, self._judge_sync, task ) # 缓存结果 dedup_key self._dedup_key_from_data(task) self.dedup_cache[dedup_key] json.dumps(result) # 发布结果到 Redis Pub/Sub通知前端 await self.redis.publish( fjudge:result:{task[task_id]}, json.dumps(result) ) self.metrics[judged] 1 async def _auto_scale_monitor(self): 动态扩缩容监控 根据队列深度自动调整 worker 数量 为什么需要动态扩缩容比赛中判题请求 是脉冲式的固定 worker 数要么在 高峰期不够、要么在低谷期浪费资源。 while True: high_len await self.redis.llen(judge:high) normal_len await self.redis.llen(judge:normal) total_queue_depth high_len normal_len current_workers self.executor._max_workers if total_queue_depth self.scale_up_threshold: # 扩容最多不超过 max_workers 的 2 倍 new_workers min(current_workers 2, self.max_workers * 2) if new_workers current_workers: self._resize_pool(new_workers) elif total_queue_depth self.scale_down_threshold: # 缩容最少不低于 max_workers 的一半 new_workers max(current_workers - 1, self.max_workers // 2) if new_workers current_workers: self._resize_pool(new_workers) await asyncio.sleep(5) # 每 5 秒检查一次 def _resize_pool(self, new_size: int): 调整线程池大小创建新池替换 old_executor self.executor self.executor ThreadPoolExecutor(max_workersnew_size) # 旧池的现有任务会继续执行完 # 新任务使用新池 old_executor.shutdown(waitFalse) def _judge_sync(self, task: dict) - dict: 同步执行判题 实际应包括编译、沙箱执行、输出比对等步骤 # 省略具体实现 return {verdict: AC} def _dedup_key(self, task: JudgeTask) - str: 生成去重键 content f{task.problem_id}:{task.source_code} return hashlib.md5(content.encode()).hexdigest() def _dedup_key_from_data(self, task: dict) - str: content f{task[problem_id]}:{task[source_code]} return hashlib.md5(content.encode()).hexdigest() def get_metrics(self) - dict: return { **self.metrics, cache_hit_rate: ( self.metrics[cache_hits] / max(self.metrics[submitted], 1) ), active_workers: self.executor._max_workers }四、极限并发下的边界保护单用户提交频率限制。即使有去重缓存恶意用户仍可能通过微小修改如加空格、改注释来绕过去重。单用户的提交频率限制如每分钟 10 次是最后一道防线。长尾问题的处理。在评测节点池中如果一个节点突然变慢如负载高、磁盘 I/O 慢它可能成为整个系统的瓶颈。健康检查和慢节点驱逐机制可以保证池中所有节点的性能一致性。消息队列的积压保护。当消息队列深度超过阈值如 10000应该触发告警并启动限流——拒绝低优先级的提交只接受比赛或关键任务的提交。数据库写入的削峰。大量判题结果的同步写入会压垮数据库。更好的做法是用批量写入每 100ms 或 100 条结果汇总后一次写入将写入 QPS 降低 100 倍。五、总结判题系统的高并发优化不是单纯的加机器而是从架构层面进行解耦和缓冲。四个核心策略构成了优化的完整蓝图消息队列做异步削峰、多级缓存做重复过滤、动态扩缩容做弹性伸缩、优先级调度做资源倾斜。这个架构的本质是用时间换空间、用异步换吞吐。消息队列允许提交请求在高峰期排队等待而不必承受瞬时冲击。当面试官问判题系统怎么支撑高并发时不要只说加机器而要说清楚消息队列的削峰原理、缓存的去重价值、以及动态伸缩的资源利用率。