FastAPI构建高并发抽奖服务：从云服务器部署到防刷实战-北京尧图网络

1. 为什么选FastApi做抽奖服务从框架选型说起先说结论抽奖服务这种业务形态用FastApi来做是我目前用下来最顺手的选择没有之一。很多人在选型的时候纠结一件事我到底是用Spring Boot、Go还是Python系框架如果只是做一个页面抽奖服务答案是看你的团队技术栈和部署环境。但如果你已经确定用Python那FastApi基本就是正确选项。原因很直接第一它天然支持异步高并发抽奖这种场景异步IO能扛住的连接数远高于同步框架第二代码量极少同样的抽奖逻辑用Flask写要一百多行FastApi五十行内搞定第三自动生成交互式API文档调试的时候省掉一大半时间。我见过不少团队用Flask搭这种小服务说实话也不是不能跑但要处理并发、参数校验、文档维护这些问题时Flask的短板就出来了。FastApi底层是Starlette跑在Uvicorn上性能上本身就比Flask高一个量级。而且抽奖服务必然要接前端页面FastApi对静态文件和模板的支持也够用不需要额外折腾。再说部署环境。标题里提到了云服务器我这次用的是阿里云的轻量应用服务器2核2G的配置带宽5M跑这样一个抽奖服务绰绰有余。学生服务器、免费试用机都可以这类服务本身不重真正的性能瓶颈在数据库连接和网络带宽不在计算能力。整个项目的最终形态是这样的用户打开一个网页点击抽奖按钮前端向后端发请求FastApi处理抽奖逻辑返回中奖结果前端弹窗展示。所有参与者名单和中奖记录都存数据库里管理端可以查看中奖列表。功能不复杂但麻雀虽小五脏俱全涉及的技术点覆盖了FastApi路由、依赖注入、数据库操作、静态文件托管、部署上线一条链路全是干货。2. 动手前的准备云服务器环境与项目骨架2.1 服务器基础环境配置云服务器拿到手之后第一步是登录然后安排环境。我用的是阿里云系统选的Ubuntu 22.04干净、稳定、Python支持好。登录服务器后先做几件事更新系统包、创建专用用户、安装基础工具。这些步骤别跳过后续能省很多坑。# 更新系统包 sudo apt update sudo apt upgrade -y # 安装基础工具 sudo apt install -y python3-pip python3-venv nginx git ufw # 开放防火墙端口SSH、HTTP、HTTPS sudo ufw allow 22 sudo ufw allow 80 sudo ufw allow 443 sudo ufw enable防火墙这里我要多说一句。很多人部署完服务发现外网访问不了排查半天才发现是防火墙没放行端口。阿里云的安全组规则和服务器内防火墙是两个层面两层都有可能需要配置。阿里云控制台里入方向规则要加一条允许TCP 80端口服务器内的ufw也要放行。两边同时放行才算真正打通。2.2 Python虚拟环境与FastApi安装这里强烈建议用虚拟环境不要直接往系统Python里装包。之前踩过坑系统自带的Python环境被弄乱之后某些系统工具直接罢工特别折腾。如果你用的是新版本Ubuntu系统自带的Python版本基本是3.10以上可以直接用venv# 创建工作目录 mkdir -p /opt/lottery cd /opt/lottery # 创建虚拟环境 python3 -m venv venv # 激活虚拟环境 source venv/bin/activate # 安装依赖 pip install fastapi uvicorn sqlalchemy pydantic python-dotenv也可以用uv这个包管理器现在挺火的创建虚拟环境比原生venv快不少pip install uv uv venv source .venv/bin/activate uv pip install fastapi uvicorn sqlalchemy pydantic python-dotenv装完之后验证一下python -c import fastapi; print(fastapi.__version__)能正常输出版本号说明环境没问题。2.3 项目目录结构设计项目结构直接决定了后续维护的体验我习惯按照功能模块划分小项目不用太复杂但也不能把所有逻辑堆在一个文件里。这次的结构是这样的/opt/lottery/ ├── app/ │ ├── __init__.py │ ├── main.py # FastApi入口 │ ├── models.py # 数据库模型 │ ├── schemas.py # Pydantic数据模型 │ ├── database.py # 数据库连接 │ ├── router/ │ │ ├── __init__.py │ │ ├── lottery.py # 抽奖接口 │ │ └── admin.py # 管理接口 │ ├── static/ │ │ ├── index.html # 抽奖页面 │ │ ├── style.css │ │ └── script.js ├── venv/ ├── requirements.txt为什么要这样拆分核心原因是抽奖接口和管理接口的逻辑不同后续扩展时不会互相干扰。比如某天要给抽奖接口加权重逻辑只需要改lottery.py不会影响到管理端。FastApi的APIRouter机制就是干这个用的天然支持模块化路由。3. 核心实现抽奖接口、数据库与页面联调3.1 数据库模型设计抽奖服务离不开数据存储。最简单的方案是用SQLite零配置、单文件、够用。但考虑到后续可能并发比较高直接用SQLAlchemy做ORM将来切换到MySQL、PostgreSQL不需要改业务代码。先看database.pyfrom sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL sqlite:///./lottery.db engine create_engine( SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False} ) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base()注意check_same_thread: False这个参数SQLite默认只允许创建它的线程访问但FastApi的异步环境下请求可能在不同线程中处理不加这个参数会报错。这是新手最容易踩的坑之一。然后是models.pyfrom sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey from sqlalchemy.sql import func from .database import Base class Prize(Base): 奖品表 __tablename__ prizes id Column(Integer, primary_keyTrue, indexTrue) name Column(String, nullableFalse) level Column(Integer, default1) # 奖品等级 total_count Column(Integer, default1) # 奖品总数 remaining_count Column(Integer, default1) # 剩余数量 probability Column(Integer, default0) # 权重 image_url Column(String, nullableTrue) # 奖品图片 created_at Column(DateTime(timezoneTrue), server_defaultfunc.now()) class Record(Base): 抽奖记录表 __tablename__ records id Column(Integer, primary_keyTrue, indexTrue) user_id Column(String, nullableFalse) # 用户标识 prize_id Column(Integer, ForeignKey(prizes.id), nullableTrue) # 中奖奖品 prize_name Column(String, nullableTrue) is_win Column(Boolean, defaultFalse) created_at Column(DateTime(timezoneTrue), server_defaultfunc.now())设计思路上有一个关键点抽奖记录表记录了每次抽奖行为哪怕没中奖也记录一笔。这样后续做概率校验、数据分析都有据可查。is_win字段和prize_id字段组合使用没中奖时prize_id为空is_win为False。3.2 抽奖核心逻辑基于权重的随机算法抽奖最核心的部分就是抽奖算法。常见的做法是固定概率比如一等奖1%、二等奖5%、三等奖10%。但业务上这种模式灵活性差改一次概率要改代码。我用的是权重模型每个奖品设一个权重值后端根据权重计算中奖概率。import random from sqlalchemy.orm import Session from ..models import Prize def draw_prize(db: Session): 基于权重的抽奖算法返回中奖的Prize对象未中奖返回None prizes db.query(Prize).filter(Prize.remaining_count 0).all() if not prizes: return None weight_sum sum(prize.probability for prize in prizes) if weight_sum 0: return None # 生成一个随机数落在哪个区间就中哪个奖 random_num random.randint(1, weight_sum) current 0 for prize in prizes: current prize.probability if random_num current: return prize return None这个算法的逻辑背后来解释一下假设有三个奖品权重分别是10、20、30权重总和是60。程序生成一个1到60的随机整数。一等奖区间是1到10二等奖区间是11到30三等奖区间是31到60。随机数落在哪个区间就对应该奖品。这种方式的好处是调整中奖概率只需要改数据库中的权重值不需要改代码、重新部署。比如某天你们做活动想要提高二等奖的中奖率直接把二等奖的权重从20改成40即时生效。但这里有个隐藏问题奖品是有剩余库存的。如果一个奖品库存为0它不应该参与抽奖所以查询时加了remaining_count 0的过滤条件。这个逻辑必须在数据库查询时处理不能在Python里筛否则并发情况下可能会把最后一件库存抽给多个用户。3.3 FastApi路由与接口实现抽奖接口用FastApi的APIRouter来组织。先看router/lottery.pyfrom fastapi import APIRouter, Depends, HTTPException, Request from sqlalchemy.orm import Session from datetime import datetime from ..database import SessionLocal from ..models import Prize, Record from ..schemas import DrawResponse from ..lottery_logic import draw_prize router APIRouter(prefix/api/lottery, tags[lottery]) def get_db(): db SessionLocal() try: yield db finally: db.close() router.post(/draw, response_modelDrawResponse) async def draw(request: Request, db: Session Depends(get_db)): 抽奖接口 # 获取用户标识 user_id request.headers.get(X-User-Id) if not user_id: raise HTTPException(status_code400, detail缺少用户标识) # 每位用户每分钟只能抽一次限制频率 last_record db.query(Record).filter( Record.user_id user_id ).order_by(Record.created_at.desc()).first() if last_record: time_diff (datetime.utcnow() - last_record.created_at.replace(tzinfoNone)).total_seconds() if time_diff 60: raise HTTPException(status_code429, detail操作太频繁请稍后再试) # 执行抽奖 prize draw_prize(db) if prize is None: # 没中奖 record Record(user_iduser_id, prize_idNone, prize_nameNone, is_winFalse) db.add(record) db.commit() return {is_win: False, prize: None, message: 很遗憾未中奖} # 中奖了扣减库存记录中奖信息 prize.remaining_count - 1 record Record(user_iduser_id, prize_idprize.id, prize_nameprize.name, is_winTrue) db.add(record) db.commit() db.refresh(prize) return { is_win: True, prize: {id: prize.id, name: prize.name, level: prize.level}, message: f恭喜你抽中了{prize.name} }这里有几个细节值得展开频率限制抽奖场景最容易遇到的问题就是刷接口。如果没有频率限制有人可以用脚本一秒发几百次请求奖品很快被薅光。我在接口里加了一个最简单的限制同一个用户标识一小时内只能抽一次。更严格的做法可以用Redis做滑动窗口限流但对这个小项目来说数据库查询实现已经够用。数据库事务扣减库存和写记录必须保证原子性。这段代码里prize.remaining_count - 1和db.add(record)以及最后的db.commit()是一个事务要么全部成功要么全部失败。如果先扣库存再提交时失败库存也不会减少因为都在同一个事务里。依赖注入db: Session Depends(get_db)是FastApi的依赖注入机制。每个请求进来时FastApi会自动从get_db中获取数据库会话请求结束时自动关闭。不需要在每个接口中手动管理数据库连接。再看router/admin.py管理端接口主要提供奖品管理功能from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from typing import List from ..database import SessionLocal from ..models import Prize, Record from ..schemas import PrizeCreate, PrizeUpdate, PrizeOut, RecordOut router APIRouter(prefix/api/admin, tags[admin]) def get_db(): db SessionLocal() try: yield db finally: db.close() router.get(/prizes, response_modelList[PrizeOut]) async def list_prizes(db: Session Depends(get_db)): 奖品列表 return db.query(Prize).order_by(Prize.level).all() router.post(/prizes, response_modelPrizeOut) async def create_prize(prize: PrizeCreate, db: Session Depends(get_db)): 新建奖品 db_prize Prize( nameprize.name, levelprize.level, total_countprize.total_count, remaining_countprize.total_count, probabilityprize.probability, image_urlprize.image_url ) db.add(db_prize) db.commit() db.refresh(db_prize) return db_prize router.get(/records, response_modelList[RecordOut]) async def list_records(db: Session Depends(get_db)): 中奖列表 return db.query(Record).filter(Record.is_win True).order_by(Record.created_at.desc()).limit(50).all()管理端这里没有做登录鉴权。如果你部署到公网建议至少加一个简单的API Key校验。可以在依赖注入中做一个全局检查要求请求头必须携带指定的Token。我在后面部署章节会讲到。3.4 Pydantic数据模型schemas.py用于定义接口的数据结构。FastApi会根据Pydantic模型自动做参数校验和响应序列化这是FastApi的一大优势代码写起来非常省心。from pydantic import BaseModel from typing import Optional, List from datetime import datetime class PrizeBase(BaseModel): name: str level: int 1 total_count: int 1 probability: int 0 image_url: Optional[str] None class PrizeCreate(PrizeBase): pass class PrizeUpdate(BaseModel): name: Optional[str] None level: Optional[int] None total_count: Optional[int] None probability: Optional[int] None image_url: Optional[str] None class PrizeOut(PrizeBase): id: int remaining_count: int created_at: datetime class Config: from_attributes True class RecordOut(BaseModel): id: int user_id: str prize_id: Optional[int] prize_name: Optional[str] is_win: bool created_at: datetime class Config: from_attributes True class DrawResponse(BaseModel): is_win: bool prize: Optional[dict] None message: strfrom_attributes True是Pydantic v2的写法允许直接从ORM对象转为响应模型。在v1版本中这个配置叫orm_mode True升级到v2后要注意修改否则会报错。3.5 前端页面与FastApi集成前端页面的实现方式有两种一是前后端分离前端单独部署到Nginx通过/api路径访问后端二是FastApi托管静态文件所有资源由后端统一服务。对于这种小项目我倾向于第二种部署起来更简单一个端口搞定所有请求。FastApi托管静态文件非常方便from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse from fastapi.middleware.cors import CORSMiddleware from .router import lottery, admin app FastAPI(title抽奖服务) # CORS配置允许前端跨域访问 app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 注册路由 app.include_router(lottery.router) app.include_router(admin.router) # 托管静态文件 app.mount(/static, StaticFiles(directoryapp/static), namestatic) app.get(/) async def root(): return FileResponse(app/static/index.html) app.on_event(startup) def startup(): # 启动时初始化数据库 from .database import Base, engine Base.metadata.create_all(bindengine) # 初始化奖品数据 init_prizes()启动时初始化数据库的操作使用Base.metadata.create_all它会自动创建所有数据表。然后调用init_prizes()在数据库空的时候插入默认的奖品数据。def init_prizes(): from .database import SessionLocal from .models import Prize db SessionLocal() try: if db.query(Prize).count() 0: default_prizes [ Prize(name一等奖iPhone 15 Pro, level1, total_count1, remaining_count1, probability1), Prize(name二等奖蓝牙耳机, level2, total_count3, remaining_count3, probability10), Prize(name三等奖定制礼品杯, level3, total_count20, remaining_count20, probability30), Prize(name谢谢参与, level0, total_count999999, remaining_count999999, probability59), ] db.add_all(default_prizes) db.commit() finally: db.close()这里注意谢谢参与的处理方式它也是一个奖品但权重占比最高这样抽奖逻辑不需要特殊处理未中奖的情况统一走奖品流程。只是前端看到这个奖品名称就知道是没中奖。前端页面index.html核心结构!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title幸运抽奖/title link relstylesheet href/static/style.css /head body div classcontainer h1幸运大抽奖/h1 div classuser-info span当前用户/span input typetext iduserId placeholder请输入用户编号 /div div classlottery-box div classprize-display idprizeDisplay span classplaceholder点击下方按钮开始抽奖/span /div button iddrawBtn立即抽奖/button /div div classresult-modal idresultModal div classmodal-content h2 idresultTitle/h2 p idresultMessage/p button idcloseModal知道了/button /div /div /div script src/static/script.js/script /body /html前端JS核心逻辑const drawBtn document.getElementById(drawBtn); const userIdInput document.getElementById(userId); const resultModal document.getElementById(resultModal); const resultTitle document.getElementById(resultTitle); const resultMessage document.getElementById(resultMessage); drawBtn.addEventListener(click, async () { const userId userIdInput.value.trim(); if (!userId) { alert(请输入用户编号); return; } drawBtn.disabled true; drawBtn.textContent 抽奖中...; try { const response await fetch(/api/lottery/draw, { method: POST, headers: { Content-Type: application/json, X-User-Id: userId } }); if (response.status 429) { alert(操作太频繁请稍后再试); drawBtn.disabled false; drawBtn.textContent 立即抽奖; return; } const data await response.json(); if (data.is_win) { resultTitle.textContent 恭喜中奖; resultMessage.textContent data.prize ? data.prize.name : data.message; } else { resultTitle.textContent 很遗憾; resultMessage.textContent 再试一次幸运之神等着你; } resultModal.style.display flex; } catch (error) { console.error(抽奖请求失败, error); alert(网络异常请稍后重试); } finally { drawBtn.disabled false; drawBtn.textContent 立即抽奖; } }); document.getElementById(closeModal).addEventListener(click, () { resultModal.style.display none; });抽奖过程中的视觉反馈也很重要给按钮加了禁用状态和倒计时文案用户体验会好很多。之前见过一些抽奖页面用户点完按钮没反应还以为是坏了其实就是前端没有做好状态反馈。4. 部署上线Nginx反向代理与Uvicorn进程管理4.1 本地先跑通再上服务器开发调试阶段本地跑起来很简单uvicorn app.main:app --reload --port 8000访问http://127.0.0.1:8000就能看到抽奖页面访问http://127.0.0.1:8000/docs可以看到自动生成的API文档。这是FastApi最爽的地方Swagger文档直接可用不需要单独维护API文档抽奖接口的参数、返回值一目了然。要特别注意一个细节本地调试时如果直接用app.mount(/static, StaticFiles(directoryapp/static))路径是相对当前工作目录。如果你在项目根目录下运行uvicorn app.main:app没问题但如果切换了运行目录路径就会出错。稳妥做法是用绝对路径import os BASE_DIR os.path.dirname(os.path.dirname(os.path.abspath(__file__))) app.mount(/static, StaticFiles(directoryos.path.join(BASE_DIR, static)), namestatic)4.2 用Supervisor管理Uvicorn进程本地跑通后部署到云服务器。直接用uvicorn app.main:app --host 0.0.0.0 --port 8000后台运行也可以但进程挂掉不会自动重启服务器重启后服务不会自动拉起。正规做法是用Supervisor做进程守护。先安装Supervisorsudo apt install -y supervisor创建配置文件/etc/supervisor/conf.d/lottery.conf[program:lottery] command/opt/lottery/venv/bin/uvicorn app.main:app --host 127.0.0.1 --port 8000 directory/opt/lottery userroot autostarttrue autorestarttrue stopasgrouptrue killasgrouptrue stderr_logfile/var/log/lottery_err.log stdout_logfile/var/log/lottery_out.log这里有个关键点Uvicorn只监听127.0.0.1不直接暴露公网。外部请求先进NginxNginx再转发给Uvicorn这样Nginx可以做静态缓存、防盗链、访问日志等比直接把后端暴露给公网安全得多。然后加载配置并启动sudo supervisorctl reread sudo supervisorctl update sudo supervisorctl status看到lottery RUNNING就说明进程管理起来了。4.3 Nginx配置与静态资源加速Nginx的配置核心是反向代理对外监听80端口把/路径的请求转发到Uvicorn监听的8000端口把/static路径的请求直接返回本地静态文件。server { listen 80; server_name _; # 静态文件由Nginx直接处理不经过FastApi location /static/ { alias /opt/lottery/app/static/; expires 7d; add_header Cache-Control public; } # API请求反向代理到FastApi location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } }expires 7d做静态资源缓存页面刷新时CSS和JS文件会走浏览器缓存接口请求才会打到后端显著降低后端压力。保存配置后重载Nginxsudo nginx -t sudo systemctl reload nginx这时候在浏览器访问服务器公网IP就能看到抽奖页面了。4.4 HTTPS配置免费SSL证书怎么弄现在的Web服务没有HTTPS基本说不过去。虽然抽奖服务本身是GET和POST接口数据不算特别敏感但浏览器现在对HTTP页面的权限限制越来越多比如navigator.geolocation、Notification API在非HTTPS环境下都会被禁用。而且很多活动页面如果被浏览器提示不安全用户体验非常差。用Certbot申请免费SSL证书几分钟搞定sudo apt install -y certbot python3-certbot-nginx sudo certbot --nginx -d your-domain.comCertbot会自动修改Nginx配置配置好HTTPS和自动续期。如果你没有域名只有IP地址Certbot是行不通的那就暂时用HTTP或者去申请一个便宜的域名。5. 上线后的真实问题并发安全与数据一致性抽奖服务上线跑起来后真正的问题才开始出现。这一章是我最想分享的内容因为很多问题是开发时根本遇不到、只有真实流量打进来才会暴露的。5.1 并发扣库存导致超卖问题第一个大问题是超卖。前面代码里扣减库存的操作是读取出奖品信息、修改remaining_count、再写回数据库。在单用户场景下没问题但如果多个请求同时到达就可能出现两个请求同时读到剩余库存为1然后同时扣减最后库存变成-1。SQLite对并发的处理尤其薄弱它的写锁是全局的多个写操作并发时会直接报database is locked错误。解决思路有三个层级第一层数据库行级锁。用SELECT FOR UPDATE锁定需要修改的行但SQLite不支持这个语法需要用BEGIN IMMEDIATE事务来锁定整个数据库。对于MySQL和PostgreSQL可以直接用。from sqlalchemy import text def draw_with_lock(db: Session, user_id: str): 使用数据库锁避免超卖 db.execute(text(BEGIN IMMEDIATE)) try: prize draw_prize(db) if prize: prize.remaining_count - 1 # 处理记录 db.commit() except: db.rollback() raise第二层应用层加锁。使用Redis分布式锁或者Python的threading.Lock。但应用层锁只对单进程有效如果将来Uvicorn起了多个worker进程锁就失效了。第三层原子更新。把判断库存是否充足和扣减库存合并成一条SQL语句from sqlalchemy import update # 原子扣减先更新再判断受影响行数 result db.execute( update(Prize) .where(Prize.id prize_id, Prize.remaining_count 0) .values(remaining_countPrize.remaining_count - 1) ) if result.rowcount 0: # 库存不足或已被抢完 pass这个做法的核心是WHERE条件里加上remaining_count 0只有满足条件时才会更新影响行数返回0就说明库存被抢完了。不需要额外加锁数据库层面的原子性就保证了不会超卖。我在项目中采用了第三种方案简单可靠也没有额外依赖。5.2 SQLite还是MySQL小型服务怎么选SQLite作为开发环境非常合适因为零配置随开随用。但生产环境如果并发量上来SQLite确实吃力。如果你预计抽奖活动参与人数不超过一万且不是大促秒杀场景SQLite完全够用。但如果预期并发较高建议直接切换到PostgreSQL或者MySQL。切换方式很简单因为用了SQLAlchemy只需要修改database.py中的连接字符串# MySQL SQLALCHEMY_DATABASE_URL mysqlpymysql://username:passwordlocalhost/lottery?charsetutf8mb4 # PostgreSQL SQLALCHEMY_DATABASE_URL postgresql://username:passwordlocalhost/lottery然后安装对应驱动即可。切换后数据表结构完全不用变业务代码也不用动。5.3 日志与监控上线后必不可少的工程化能力服务上线后最怕遇到的情况是用户说抽不了奖但你不知道是什么原因。所以日志非常重要。Uvicorn本身就输出访问日志但默认只打印到控制台。用Supervisor配置时已经加了日志文件路径stdout_logfile/var/log/lottery_out.log stderr_logfile/var/log/lottery_err.log如果接口报错错误信息会写入lottery_err.log排查问题时直接tail -f /var/log/lottery_err.log在FastApi代码里自己也可以加一些关键日志import logging logger logging.getLogger(lottery) logger.setLevel(logging.INFO) router.post(/draw) async def draw(request: Request, db: Session Depends(get_db)): logger.info(f用户 {user_id} 发起抽奖) try: # 抽奖逻辑 logger.info(f用户 {user_id} 中奖{prize.name}) except Exception as e: logger.error(f用户 {user_id} 抽奖异常{str(e)}) raise日志是定位问题的第一手段一定要养成写日志的习惯。很多小项目上线出了问题第一个动作是登录服务器看一眼日志没有日志就只能干瞪眼。我见过一个活动服务因为没打日志接口报错两天没发现用户投诉了一堆才发现问题非常被动。5.4 防刷抽奖服务必须要做的安全措施抽奖服务的防刷是重中之重。不设防的话有人拿脚本刷你接口奖品分分钟被薅完。这个项目里做了三道基础防护第一道用户标识限制。每个请求必须携带X-User-Id头后端记录每个用户的抽奖次数和频率。代码里已经实现了同一用户一分钟内只能抽一次的限制。第二道全局频率限制。用中间件限制单个IP的请求频率。简单做法是在Nginx层做限制limit_req_zone $binary_remote_addr zonelottery_limit:10m rate5r/s; server { location /api/lottery/draw { limit_req zonelottery_limit burst10 nodelay; proxy_pass http://127.0.0.1:8000; } }这样任何一个IP每秒最多5次请求超出部分直接返回503。对正常用户没影响因为正常用户不可能每秒点5次抽奖按钮。第三道验证码。如果你做的是一次大活动奖品价值较高强烈建议在抽奖请求前加滑块验证码或人机校验。前后端都需要配合前端弹出验证码通过后获取一个token后端校验token有效才允许抽奖。三道防护叠加后刷接口的成本会大幅提升。虽然不能绝对杜绝有耐心的攻击者但至少能挡住99%的脚本小子。6. 从开发到上线最容易踩的坑实测排查记录6.1 阿里云安全组忘记开端口这是我最想提醒新手的坑。本地服务已经跑通了但浏览器访问公网IP总是超时。在服务器上curl http://127.0.0.1:8000能通说明后端进程正常但用手机流量访问公网IP就不通。排查步骤确认Uvicorn监听地址是0.0.0.0不是127.0.0.1。如果你用--host 127.0.0.1启动外部当然无法访问。确认Nginx进程正常systemctl status nginx。确认服务器防火墙放行ufw status。最后才是安全组登录阿里云控制台找到ECS安全管理查看安全组规则。当时我的情况是前3步都没问题问题出在安全组。阿里云轻量应用服务器的控制台里防火墙规则需要手动添加80端口。默认只开放了22、443等少数端口80端口居然没开被这个坑卡了半小时。6.2 SQLite数据库文件权限问题部署到服务器后第一次抽奖请求就报错错误日志显示OperationalError: unable to open database file原因是SQLite数据库文件lottery.db是运行uvicorn的root用户创建的但Supervisor配置中如果指定了其他用户运行或者Nginx进程中某些操作需要访问数据库文件就会因权限不足而报错。解决方法是统一运行用户并把数据库文件放到有权限的目录chown root:root /opt/lottery chmod 755 /opt/lottery最简单的做法是让数据库文件、项目目录全部归同一个用户配置文件里也统一用这个用户运行。6.3 FastApi的on_event(startup)被弃用问题FastApi新版本已经不建议使用app.on_event(startup)而是推荐用lifespan上下文处理器。如果你在最新的FastApi版本中继续用on_event控制台会输出DeprecationWarning。最新的写法from contextlib import asynccontextmanager asynccontextmanager async def lifespan(app: FastAPI): # 启动时执行 print(服务启动初始化数据库...) Base.metadata.create_all(bindengine) init_prizes() yield # 关闭时执行 print(服务关闭清理资源...) app FastAPI(title抽奖服务, lifespanlifespan)新版写法在生命周期管理上更清晰启动逻辑和关闭逻辑放在同一个上下文处理器里不会遗漏。6.4 Pydantic v2兼容性from_attributesvsorm_mode做管理端接口时返回奖品列表一直报错AttributeError: Prize object has no attribute from_attributes排查半天才发现是Pydantic v1和v2的兼容问题。Pydantic v2中Config类里的配置项改名了# Pydantic v1 class Config: orm_mode True # Pydantic v2 class Config: from_attributes True如果你在使用较新的FastApi版本0.100.0以上它已经依赖Pydantic v2配置项必须是from_attributes。老项目升级时很容易掉进这个坑。6.5 数据库连接泄漏上线跑了一阵子后服务突然变慢日志出现大量数据库连接错误。排查发现是数据库连接没有正确释放。问题出在早期版本中我在接口内部直接创建了SessionLocal()但忘记关闭。正确的做法是用FastApi的依赖注入机制让get_db在每个请求结束时自动关闭连接。使用依赖注入后这个问题就不存在了。所以写FastApi时一定要养成依赖注入的习惯不要自己手动管理连接生命周期否则很容易泄漏。7. 进阶优化从能用到好用基础版本上线后如果你想让这个抽奖服务更像一个正经产品有几个方向可以继续深入。7.1 抽奖大屏模式活动场景下往往需要一个投影大屏实时滚动展示中奖名单。这个功能可以在现有基础上加一个SSE或者WebSocket推送接口每次有人中奖服务端主动推送给大屏页面。FastApi对WebSocket的支持很完善from fastapi import WebSocket, WebSocketDisconnect class ConnectionManager: def __init__(self): self.active_connections: list[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast(self, message: str): for connection in self.active_connections: await connection.send_text(message) manager ConnectionManager() router.websocket(/ws/records) async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: await websocket.receive_text() except WebSocketDisconnect: manager.disconnect(websocket)抽奖接口中奖后调用manager.broadcast(json.dumps(record))大屏通过WebSocket收到新消息自动滚动展示。7.2 奖品权重动态调整活动运营中经常需要临时调整概率。比如前200名参与用户中奖率翻倍或者某个时段特等奖概率临时提高。目前的设计中奖品权重存在数据库里运营人员可以直接改数据库字段或者提供一个后台管理接口。如果是更精细化的运营策略可以引入时间段维度在奖品表中增加start_time和end_time字段抽奖逻辑根据当前时间判断走哪套概率。7.3 用户中奖次数限制有些活动会限制同一个用户的总中奖次数比如每人最多中一次一等奖。在draw_prize抽之前先查询该用户的中奖记录# 检查用户是否已中过一等奖 won_first_prize db.query(Record).filter( Record.user_id user_id, Record.is_win True, Record.prize_name 一等奖iPhone 15 Pro ).count() if won_first_prize 0: return None # 或者返回一个提示注意如果用户已中过一等奖需要单独提示已获得过一等奖不能重复参与而不是简单返回未中奖。前端要根据后端返回的message字段做不同展示。7.4 数据统计与活动复盘活动结束后运营团队一定想要一份完整的数据报告参与人数、总抽奖次数、中奖率、各奖项兑奖率。目前的数据表结构已经支持这些统计查询router.get(/api/admin/stats) async def stats(db: Session Depends(get_db)): total_draws db.query(Record).count() total_wins db.query(Record).filter(Record.is_win True).count() win_rate (total_wins / total_draws * 100) if total_draws 0 else 0 # 按奖品维度统计 prize_stats db.query(Record.prize_name, func.count(Record.id)).filter( Record.is_win True ).group_by(Record.prize_name).all() return { total_draws: total_draws, total_wins: total_wins, win_rate: round(win_rate, 2), prize_stats: [{name: name, count: count} for name, count in prize_stats] }数据报表做出来之后运营可以清晰看到每种奖品的中奖情况对活动效果有明确认知。这套服务从开发到部署跑通最后上线稳定运行整个链路其实并不复杂。FastApi的简洁设计和丰富文档让开发体验非常好部署层面用Nginx加Supervisor又是非常成熟的方案。关键是过程中要把数据库并发安全、接口防刷、日志监控这些看不见但很致命的细节考虑进去而不是只管接口能通、页面能看就完事了。我个人实际操作中的最大体会是技术栈选型不是最难的难的是上线前把各种边界场景想清楚。抽奖服务这块你多花一小时做防刷和并发处理上线后就能少熬几个通宵。

FastAPI构建高并发抽奖服务：从云服务器部署到防刷实战

相关推荐

破碎机振动超标诊断与治理：从频谱分析到弹簧隔振的完整实践

推荐算法实战：行为日志驱动的用户购买预测与XGBoost调优

基于 Python 的宠物食品销售数据分析可视化系统的设计与实现

更多资讯

智能体架构演进：从ReAct到LangGraph的技术解析

电力系统N-k安全约束下光热电站优化调度模型

Dapr Workflow 性能测试报告解读：v1.16.3 工作流压测图表手册

写论文开一堆AI会员还总踩坑？2026从开题到降重按这分工就对了

ASP.NET MVC在线考试系统源码解析：路由、控制器与计分全流程

Python实现替换密码破解：频率分析与优化策略

今日更新

AI驱动的自动化专利分析平台技术解析

LeetCode-Go 题解：1252. Cells with Odd Values in a Matrix（奇数值单元格计数）双解法剖析

在线答疑系统Java毕设实战：Spring Boot前后端分离与状态流转

本周精选

AI驱动的自动化专利分析平台技术解析

LeetCode-Go 题解：1252. Cells with Odd Values in a Matrix（奇数值单元格计数）双解法剖析

在线答疑系统Java毕设实战：Spring Boot前后端分离与状态流转

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？