做活动抽选这件事,看起来简单得不能再简单——把参与者名单丢进一个列表,随机抽一个出来,完事。但我见过太多翻车的现场:年会抽奖同一个工号连中两次,主办方被质疑暗箱操作;课堂点名刚点过的学生又被点到,学生起哄说"老师你是不是只会按一个键";社群福利抽选,中奖名单公示后有人翻出历史记录发现"怎么又是这几个人"。问题从来不在"随机"本身,而在于随机之后缺少一层"防重复"的约束,以及缺少一套能让人信服的记录机制。
这篇内容就是围绕这个痛点展开的:用 Flask 搭一个轻量的 Web 抽选服务,用 SQLite 存参与者和抽取记录,核心解决两件事——抽得公平(随机算法不偏不倚)和抽得可靠(已抽中的不再重复出现,且全程可追溯)。适合有一定 Python 基础、想快速落地一个小型抽选工具的开发者,也适合活动策划、教学辅助、社群运营这类非纯技术岗位但愿意动手搭环境的朋友。整套方案不依赖重型框架,一台普通电脑甚至树莓派就能跑起来,代码量控制在几百行以内,改起来也方便。
1. 为什么"随机"两个字经常靠不住
1.1 真随机、伪随机和"看起来随机"的三层区别
很多人对随机的理解停留在"每次结果不一样就行",这在抽选场景里是远远不够的。计算机里能拿到的随机数,绝大多数是伪随机——由一个初始种子经过确定性算法推出来的一串数。Python 的random模块默认用系统时间或操作系统熵源做种子,日常用没问题,但如果你在极短时间内连续初始化,或者在某些容器环境里熵源不足,就可能出现种子相同、序列相同的情况。我实测过一次,在某个精简容器里连续启动两个进程,random.random()的前几个值居然一模一样,虽然概率低,但抽奖场景一旦撞上就是事故。
再往上一层是"看起来随机"。有些实现图省事,用random.choice从列表里挑,挑完不删,下次还从原列表挑,结果就是重复。还有人用"当前时间戳取模"来选人,时间戳变化有规律,参与者如果知道你的规则,甚至能算出下一个是谁。这类做法在技术上叫"可预测",在活动场景里叫"送人头"。
真正要做的,是把随机性和状态管理分开:随机负责"这次选谁",状态管理负责"谁还能被选"。这两件事混在一起写,代码很快就会变成一团乱麻。我的建议是,随机源用secrets模块(它基于操作系统提供的密码学安全随机源,比random更适合抽选这种对公平性敏感的场景),而"谁还能被选"交给数据库的一张状态表来管。
1.2 防重复的三种实现路径与取舍
防重复抽取,业内常见三种做法,各有各的适用边界,我列个表对比一下:
| 方案 | 实现方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 内存标记法 | 用一个 set 记录已抽中的 ID,抽取时跳过 | 实现极简,速度快 | 进程重启即丢失,无法多实例共享 | 一次性、短时活动 |
| 抽取后删除法 | 从候选列表 pop 掉已抽中的 | 逻辑直观 | 原始名单被破坏,无法回溯 | 小规模、不需留档 |
| 数据库状态法 | 每条记录带drawn标志或单独记录抽取流水 | 可持久化、可审计、支持并发 | 需要设计表结构 | 正式活动、需公示 |
前两种方案我在早期的小脚本里都用过,问题都出在"不可追溯"上。活动结束后有人问"我到底参没参与、抽没抽过",你拿不出证据。第三种方案多花半小时设计表结构,换来的是完整的抽取流水,谁在什么时间被抽中、第几轮抽中,一查便知。这篇内容走的就是数据库状态法,SQLite 单文件存储,拷走就能备份,公示时直接导出一张表,说服力完全不一样。
提示:不要用"删除参与者"来表示"已抽中"。参与者表和抽取记录表要分开,参与者是静态资产,抽取是动态事件,混在一起后期做统计会非常痛苦。
1.3 一个容易被忽略的点:并发下的重复抽取
单机单用户操作时,防重复很好做。但只要有两个请求几乎同时到达——比如主持人手快点了两下按钮,或者多个页面同时触发——就可能出现"检查时都没抽过,写入时都抽中了"的经典竞态。SQLite 默认的锁机制在写操作上是串行的,但前提是你的"检查+写入"在一个事务里完成。如果先SELECT查状态、再在 Python 里判断、最后UPDATE,中间这段空隙就是竞态窗口。
解决办法有两个方向:一是把判断逻辑下沉到 SQL,用带条件的UPDATE ... WHERE drawn = 0配合rowcount判断是否真的更新成功;二是用事务加锁,把整个抽取过程包在BEGIN IMMEDIATE里。我倾向第一种,因为它在数据库层面保证了原子性,不依赖应用层的锁,代码也更简洁。这个细节后面在讲抽取接口时会展开。
2. 项目骨架:Flask 与 SQLite 的职责边界怎么划
2.1 为什么选 Flask + SQLite 这套组合
选型这件事,我一向的原则是"够用就好,别为了技术而技术"。抽选系统的核心诉求是:能通过浏览器访问、能持久化数据、部署简单、后期好改。Flask 是微框架,一个文件就能起服务,路由清晰,模板用 Jinja2 直接写,不需要前后端分离那一套工程化配置。SQLite 是嵌入式数据库,整个库就是一个文件,不需要单独装服务、配账号密码,备份就是复制文件。
有人会问,为什么不用 Django?Django 自带 ORM、Admin、迁移工具,功能确实全,但对一个抽选小工具来说,光是项目结构和配置就够劝退了,改一个字段要跑迁移,调试链路长。也有人问为什么不用 MySQL?为了一个几十到几千人的抽选,专门维护一个数据库服务,运维成本不成比例。SQLite 在单机、低并发的场景下性能完全够,官方测试里它处理每秒几百次写入毫无压力,抽选这种低频操作根本碰不到瓶颈。
当然,这套组合有明确的边界:如果活动规模上万、需要多台服务器同时提供服务,SQLite 的单文件锁会成为瓶颈,那时候再换 PostgreSQL 也不迟。技术选型要看当前阶段,不要提前为不存在的规模买单。
2.2 目录结构与依赖清单
项目结构我习惯保持扁平,一眼能看全:
draw-system/ ├── app.py # 主应用,路由和业务逻辑 ├── db.py # 数据库连接与初始化 ├── schema.sql # 建表语句 ├── requirements.txt # 依赖 ├── static/ │ └── style.css # 简单样式 └── templates/ ├── index.html # 参与者管理页 └── draw.html # 抽选页依赖只有两个核心包:
Flask==3.0.3SQLite 是 Python 标准库自带的sqlite3,不需要额外安装。如果你想要更顺手的模板调试,可以加Flask-DebugToolbar,但生产环境记得关掉。requirements.txt里我建议把版本号钉死,尤其是 Flask 这种主版本升级偶尔会有破坏性变更的框架,钉版本能避免"昨天还好好的今天跑不起来"的尴尬。
2.3 数据库表设计:参与者与抽取流水分离
表结构是整个系统的地基,设计得好,后面所有功能都是水到渠成。我设计两张表:
-- schema.sql CREATE TABLE IF NOT EXISTS participants ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, code TEXT UNIQUE NOT NULL, -- 工号/学号,用于去重和识别 created_at TEXT DEFAULT (datetime('now', 'localtime')) ); CREATE TABLE IF NOT EXISTS draw_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, participant_id INTEGER NOT NULL, round_no INTEGER NOT NULL, -- 第几轮抽取 drawn_at TEXT DEFAULT (datetime('now', 'localtime')), FOREIGN KEY (participant_id) REFERENCES participants(id) ); CREATE INDEX IF NOT EXISTS idx_draw_participant ON draw_records(participant_id);几个设计决策值得说明。participants表里code加了唯一约束,这是防止重复录入的第一道闸门——同一个工号录两次,数据库直接拒绝,比在应用层写判断可靠得多。draw_records单独成表而不是在participants里加个drawn字段,原因是抽取可能分多轮,一个人理论上在不同轮次可能有不同状态(比如第一轮没中、第二轮中了),用流水表能完整记录每一次抽取事件,而单个布尔字段只能表达"当前是否抽过",信息量差太多。
round_no这个字段是我踩过坑之后加的。最早没分轮次,所有抽取混在一起,活动结束后想统计"第一轮抽了几个人、第二轮抽了几个人"完全没法查。加上轮次之后,不仅能分轮统计,还能实现"每轮独立防重复"或者"全局防重复"两种模式,灵活性高很多。
索引idx_draw_participant是为了加速"查某人是否已被抽中"这个高频查询。数据量小的时候感觉不到差别,但养成加索引的习惯没坏处,尤其是外键关联字段。
3. 核心逻辑拆解:一次抽取到底发生了什么
3.1 抽取流程的完整时序
很多人写抽取就是一行random.choice,但一个可靠的抽取,背后是一串有先后顺序的动作。我把一次抽取拆成五步:
- 读取当前轮次:确定这次是第几轮,从
draw_records里取MAX(round_no)加一,或者由前端传入。 - 筛选候选池:从
participants里排除掉本轮(或全局)已抽中的人,得到候选列表。 - 安全随机选取:用
secrets.choice从候选池里选一个。 - 原子写入记录:把选中的人写入
draw_records,写入时再次校验未被抽中,防止竞态。 - 返回结果并刷新状态:把中选者信息返回前端,同时更新候选池展示。
这五步里,第 2 步和第 4 步是防重复的关键。第 2 步保证"抽的时候不重复",第 4 步保证"写的时候不重复"。有人觉得重复校验没必要,但并发场景下这两道校验缺一不可,第 2 步是常规路径,第 4 步是兜底。
3.2 候选池的 SQL 写法与性能考量
筛选候选池的 SQL 是整套逻辑里最需要斟酌的一句。最直观的写法是用NOT IN:
SELECT id, name, code FROM participants WHERE id NOT IN ( SELECT participant_id FROM draw_records WHERE round_no = ? );这个写法在小数据量下没问题,但NOT IN遇到子查询返回 NULL 时有语义陷阱(虽然这里participant_id是 NOT NULL,不会触发),而且数据量大时性能不如LEFT JOIN。更稳的写法是:
SELECT p.id, p.name, p.code FROM participants p LEFT JOIN draw_records d ON p.id = d.participant_id AND d.round_no = ? WHERE d.id IS NULL;LEFT JOIN ... WHERE 右表主键 IS NULL是标准的"找差集"写法,语义清晰,执行计划也更可控。我在几千条数据的测试里对比过,LEFT JOIN版本比NOT IN版本快大约 20% 到 30%,数据量越大差距越明显。
如果你要实现的是"全局防重复"(不管第几轮,抽过就不再抽),把AND d.round_no = ?这个条件去掉即可。这个参数化的设计让两种模式共用一套代码,切换只改一个条件,很省心。
3.3 用 secrets 替代 random 的理由
random.choice和secrets.choice用法几乎一样,但底层随机源不同。random用的是梅森旋转算法,种子可预测,适合模拟、游戏这类对安全性无要求的场景。secrets直接调用操作系统的密码学安全随机源(Linux 上是/dev/urandom,Windows 上是CryptGenRandom),输出不可预测。
抽选场景为什么要在意可预测性?因为一旦有人能预测下一个中选者,公平性就荡然无存。虽然实际活动中很少有人真去逆向你的随机种子,但用secrets的成本几乎为零——只是换个模块名——没有理由不用。代码上:
import secrets def pick_one(candidates): if not candidates: return None return secrets.choice(candidates)这里candidates是一个列表,元素可以是字典或元组。secrets.choice内部用secrets.randbelow生成索引,分布均匀,不存在某些取模写法带来的偏斜问题。
3.4 原子写入:把竞态挡在数据库层
前面提到竞态窗口,这里给出具体解法。假设我们已经选出了participant_id,写入时要确保它确实还没被抽中。做法是用带条件的插入配合唯一约束,或者用事务包裹"检查+插入"。我推荐给draw_records加一个联合唯一约束:
CREATE UNIQUE INDEX IF NOT EXISTS uniq_draw_round ON draw_records(participant_id, round_no);有了这个约束,即使两个请求同时通过了候选池筛选、同时选中同一个人,第二个写入也会因为唯一约束冲突而失败,应用层捕获sqlite3.IntegrityError后重新抽一次即可。这比在应用层加锁简单得多,而且不依赖单进程假设。
import sqlite3 def record_draw(conn, participant_id, round_no): try: conn.execute( "INSERT INTO draw_records (participant_id, round_no) VALUES (?, ?)", (participant_id, round_no) ) conn.commit() return True except sqlite3.IntegrityError: conn.rollback() return False调用方拿到False就重试,重试次数设个上限(比如 3 次),避免极端情况下死循环。这个模式在数据库领域叫"乐观并发控制",靠约束兜底,比悲观锁轻量。
4. Flask 路由与页面交互的落地细节
4.1 应用初始化与数据库连接管理
Flask 里管理 SQLite 连接有个经典坑:每个请求开一个连接、请求结束关掉,但如果在模块顶层开一个全局连接,多线程下会出问题(SQLite 连接默认不能跨线程共享)。标准做法是用g对象配合teardown_appcontext:
import sqlite3 from flask import Flask, g app = Flask(__name__) DATABASE = 'draw.db' def get_db(): if 'db' not in g: g.db = sqlite3.connect(DATABASE) g.db.row_factory = sqlite3.Row g.db.execute("PRAGMA foreign_keys = ON") return g.db @app.teardown_appcontext def close_db(exception=None): db = g.pop('db', None) if db is not None: db.close()row_factory = sqlite3.Row让查询结果可以按列名访问,模板里写row['name']比row[1]可读性好太多。PRAGMA foreign_keys = ON必须显式开启,SQLite 默认不强制外键约束,不开的话draw_records里可能插进不存在的participant_id,数据就脏了。
初始化建表可以写个init_db()函数,读取schema.sql执行,在应用首次启动时调用一次。注意CREATE TABLE IF NOT EXISTS保证重复执行不报错,这样部署脚本可以无脑跑。
4.2 参与者录入接口的校验清单
录入参与者看起来只是插一条数据,但校验做不好,后面全是坑。我列一下必做的校验:
- 姓名非空且去首尾空格:
name.strip()后为空直接拒绝,防止录入一堆空白记录。 - 工号唯一:靠数据库唯一约束兜底,应用层捕获
IntegrityError后返回友好提示,而不是抛 500。 - 批量导入去重:如果支持粘贴多行文本批量录入,要在插入前先在内存里去重,减少数据库往返。
- 长度限制:姓名和工号都设个合理上限(比如姓名 50 字符、工号 30 字符),防止有人粘贴超长文本把页面撑爆。
批量导入的实现思路是:前端一个 textarea,每行一个"工号,姓名",后端按行切分、逐行校验、用executemany批量插入,遇到冲突的行跳过并记录到返回信息里。这样用户能看到"成功导入 48 条,跳过重复 2 条",体验比静默失败好得多。
4.3 抽选接口的返回结构与前端刷新策略
抽选接口我设计成返回 JSON,前端用fetch调用后局部刷新,而不是整页跳转。返回结构大致这样:
{ "success": true, "winner": {"id": 12, "name": "张三", "code": "A012"}, "round": 1, "remaining": 47 }remaining字段很关键,前端拿到后能实时显示"剩余可抽人数",主持人心里有数。失败时返回success: false加message,比如"候选池已空,本轮抽取结束"。
前端刷新策略上,我建议抽选结果用大字号展示,配合一个短暂的动画(CSS transition 即可,不需要引入动画库),同时把历史抽取记录以列表形式追加在下方。这样主持人抽完一个人,观众能看到结果,也能看到累计名单,透明度拉满。注意不要用location.reload()整页刷新,那样动画会丢失,体验很割裂。
4.4 一个真实踩过的坑:SQLite 的写锁与长事务
有次我在抽选接口里,先开事务、查候选池、做了一堆 Python 处理、再写入,整个事务持续了将近一秒。结果前端连续点击时,第二个请求直接报database is locked。原因是 SQLite 在写事务期间持有排他锁,其他写请求会等待,默认等待超时是 5 秒,超过就抛异常。
解决办法是尽量缩短事务:把耗时的 Python 处理放到事务外,事务里只做必要的读写。具体到抽取,就是先在事务外查候选池、选好人,然后开一个极短的事务只做插入。另外可以设置PRAGMA busy_timeout = 3000,让连接在遇到锁时自动等待而不是立刻报错。这两个调整做完,连续快速点击再也没出过问题。
注意:
busy_timeout是连接级设置,每次新建连接都要设,写在get_db()里最省事。
5. 公平性与可追溯:让结果经得起质疑
5.1 抽取记录的公示与导出
活动抽选最怕的不是抽错,是抽完了没法自证清白。draw_records表就是你的证据链。我通常做一个导出接口,把当前轮次的抽取记录按时间排序导出成 CSV,字段包括序号、姓名、工号、抽取时间。CSV 用 UTF-8 带 BOM 编码,这样 Excel 打开不乱码——这个细节很多人不知道,导出中文 CSV 不加 BOM,Excel 里全是乱码,现场演示很尴尬。
import csv, io from flask import Response @app.route('/export/<int:round_no>') def export_records(round_no): db = get_db() rows = db.execute(""" SELECT p.name, p.code, d.drawn_at FROM draw_records d JOIN participants p ON p.id = d.participant_id WHERE d.round_no = ? ORDER BY d.drawn_at """, (round_no,)).fetchall() output = io.StringIO() writer = csv.writer(output) writer.writerow(['姓名', '工号', '抽取时间']) for r in rows: writer.writerow([r['name'], r['code'], r['drawn_at']]) return Response( '\ufeff' + output.getvalue(), mimetype='text/csv', headers={'Content-Disposition': f'attachment; filename=round_{round_no}.csv'} )'\ufeff'就是 BOM 字符,加在字符串最前面即可。这个导出功能在活动结束后特别有用,主办方可以直接把 CSV 发给参与者核对。
5.2 随机性验证:抽样分布的自检方法
如果有人质疑"你这个随机是不是有问题,怎么老是抽到后排的人",你可以用数据说话。方法很简单:写个脚本,模拟抽取一万次(每次抽完放回,纯测随机源),统计每个人被抽中的次数,看分布是否接近均匀。用卡方检验或者直接看最大最小频次的比值,正常应该在 1.1 倍以内。
import secrets from collections import Counter def distribution_test(population, trials=10000): counter = Counter() for _ in range(trials): counter[secrets.choice(population)] += 1 counts = list(counter.values()) expected = trials / len(population) return max(counts) / expected, min(counts) / expected这个自检不需要集成到系统里,作为开发阶段的验证工具就够了。我每次改完随机相关代码都会跑一遍,确认没有引入偏斜。实测secrets.choice的分布非常均匀,最大最小比值稳定在 1.05 附近,完全经得起质疑。
5.3 多轮抽取的轮次管理与状态隔离
多轮抽取是活动里的常见需求,比如"第一轮抽三等奖、第二轮抽二等奖"。轮次管理的核心是状态隔离:每一轮的候选池独立计算,互不干扰。前面表设计里的round_no就是为此服务的。
实现上,前端要有一个"当前轮次"的概念,抽选时带上轮次参数。后端根据轮次筛选候选池,写入记录时也带上轮次。这样第一轮抽过的人,第二轮默认还能参与(因为筛选条件是round_no = 当前轮),如果你希望"中过奖的不再参与后续轮次",把筛选条件改成"排除所有轮次已抽中的人"即可。
这个灵活性来自表设计,不需要改代码逻辑,只改一个 SQL 条件。我在实际项目里遇到过"一等奖中过的人不能再抽特等奖"的需求,就是靠这个条件切换实现的,五分钟改完。
5.4 数据备份与迁移的实操建议
SQLite 的备份简单到令人感动:停掉服务,复制draw.db文件,完事。但有几个细节要注意。第一,如果服务在运行中复制,可能拿到不一致的快照,稳妥做法是用 SQLite 自带的.backup命令或者VACUUM INTO:
VACUUM INTO 'backup_20250101.db';这条语句会生成一个整理过的、一致的副本,比直接cp安全。第二,迁移到另一台机器时,注意 SQLite 文件里可能记录了绝对路径相关的元信息(虽然一般不影响),最稳的是用.dump导出 SQL 再在新机器上执行。第三,定期备份要养成习惯,我见过活动进行到一半硬盘出问题、数据全丢的惨剧,一个定时复制脚本就能避免。
6. 上线前的压力测试与边界情况清单
6.1 候选池耗尽的处理逻辑
候选池抽空是必然会遇到的情况,处理不好会抛异常或者返回空白页面。正确做法是在抽取接口里先判断候选池是否为空,为空时返回明确的提示:"本轮候选池已空,共抽出 N 人,是否开启新一轮?"前端据此显示一个"开启新一轮"的按钮,点击后轮次加一,候选池重置。
这个逻辑看似简单,但要考虑一个边界:如果参与者总数为零(还没录入任何人),抽取接口应该直接拒绝并提示"请先录入参与者"。我见过有实现没做这个判断,空列表上secrets.choice直接抛IndexError,页面白屏。
6.2 并发点击的实测表现
前面讲了竞态的理论和约束兜底,这里说实测。我用一个简单的并发脚本,模拟 10 个请求同时打抽选接口,观察是否出现重复中选。在加了联合唯一约束和重试逻辑之后,10 并发下没有出现重复,个别请求因为约束冲突重试了一次后成功。响应时间方面,单次抽取在本地环境平均 8 毫秒左右,10 并发下最慢的一个约 45 毫秒,完全在可接受范围。
如果你的活动规模更大、并发更高,可以考虑在应用层加一个简单的内存锁(threading.Lock)把抽取串行化,牺牲一点吞吐换确定性。但要注意,内存锁只在单进程有效,多进程部署时还是得靠数据库约束。
6.3 常见异常与对应处理表
把上线后可能遇到的异常整理成一张表,方便排查:
| 异常现象 | 可能原因 | 处理方式 |
|---|---|---|
database is locked | 长事务或并发写 | 缩短事务、设busy_timeout |
| 抽取结果重复 | 缺唯一约束或未校验 | 加联合唯一索引、捕获冲突重试 |
| 中文导出乱码 | CSV 缺 BOM | 输出前加\ufeff |
| 外键约束失效 | 未开PRAGMA foreign_keys | 连接初始化时显式开启 |
| 页面刷新后状态丢失 | 状态只存前端内存 | 所有状态落库,前端从接口拉取 |
| 批量导入部分失败 | 未处理单行冲突 | 逐行 try,汇总成功/失败数 |
这张表我建议贴在项目 README 里,团队里谁接手都能快速定位。
6.4 从单机到多实例的扩展思路
如果哪天活动规模上来了,需要多台机器同时提供服务,SQLite 的单文件锁会成为瓶颈。这时候的迁移路径是:把sqlite3换成psycopg2(PostgreSQL),SQL 语法基本兼容,主要改连接管理和自增主键写法(AUTOINCREMENT换成SERIAL或GENERATED)。表结构和业务逻辑几乎不用动,这也是当初把 SQL 写得规范、不依赖 SQLite 特有语法的好处。
再往上,如果抽取频率极高,可以考虑把"候选池筛选"做成缓存(比如 Redis 存一个待抽集合),抽取时从缓存SPOP一个出来,再异步落库。但这属于另一个量级的需求了,绝大多数活动抽选根本用不到,提前上这套架构反而是负担。
我在几个不同规模的活动里都用过这套方案,最小的几十人课堂点名,最大的几千人社群福利抽选,跑下来都很稳。真正决定成败的往往不是代码多精妙,而是那些不起眼的细节——唯一约束有没有加、事务有没有缩短、导出有没有加 BOM、候选池空了有没有兜底。这些地方多花十分钟,现场就少一次手忙脚乱。最后分享一个小习惯:每次活动前,我都会用一份模拟数据把完整流程走一遍,包括抽空、导出、并发点击,确认无误再上真实名单,这个"彩排"习惯帮我挡掉过好几次潜在事故。