二类疫苗管理系统从零搭建:新手避坑指南与实战拆解
二类疫苗管理系统从零搭建:新手避坑指南与实战拆解
面试被问原理答不上来，代码写出来却跑不通，这是很多刚入行或者转行做后端的新手最头疼的事。很多人以为写个增删改查就是开发，真到了实战项目里，才发现权限控制、数据一致性、并发安全全是坑。今天咱们不整虚的，直接上手做一个【二类疫苗】预约与库存管理系统的核心模块。
这不是一个玩具代码，而是一个能跑通业务逻辑、具备基本高可用特性的实战案例。我会把代码拆碎了揉烂了讲，专门针对【新手避坑】场景，告诉你为什么这么写，哪里容易炸，以及生产环境里该怎么兜底。
项目目标与核心痛点
做【二类疫苗】管理系统，表面看是管理药品库存，实则是在处理“高并发下的资源竞争”和“复杂状态机流转”。
为什么选这个场景？业务闭环短：包含预约、锁定库存、支付（模拟）、核销、退款五个核心状态，逻辑清晰。
痛点集中：疫苗有有效期，库存有限，用户可能重复点击。这些都是面试高频考点。
技术覆盖面广：涉及数据库事务、乐观锁、Redis预扣减、消息队列异步处理。新手最容易踩的三个坑：坑一：直接查库扣库存。高并发下，两个用户同时读到库存为1，都执行stock = stock - 1，结果库存变成-1，或者超卖。
坑二：状态机混乱。用户取消预约后，库存没释放；或者支付超时后，库存状态卡死。
坑三：缺乏幂等性。网络抖动导致重复请求，同一用户预约了两次，库存被扣两次。我们的目标，就是构建一个能抵御这些坑的轻量级系统。技术栈选用 Python + FastAPI + PostgreSQL + Redis。为什么选Python？因为语法简洁，适合快速验证逻辑；为什么选PostgreSQL？因为它对事务和约束的支持比MySQL更严格，适合做数据一致性测试。
目录结构设计
工程化不仅仅是写代码，更是管理代码。一个合格的实战项目，目录结构必须清晰。以下是我们推荐的结构：
vaccine-system/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 入口
│ ├── config.py # 配置管理
│ ├── models/
│ │ ├── __init__.py
│ │ ├── vaccine.py # 疫苗库存模型
│ │ └── order.py # 预约订单模型
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── vaccine.py # Pydantic 数据校验
│ │ └── order.py
│ ├── services/
│ │ ├── __init__.py
│ │ ├── inventory.py # 库存核心逻辑
│ │ └── order.py # 订单业务逻辑
│ ├── repositories/
│ │ ├── __init__.py
│ │ ├── db.py # 数据库连接
│ │ └── redis_client.py # Redis 连接
│ └── utils/
│ ├── __init__.py
│ └── logger.py # 日志配置
├── tests/
│ ├── __init__.py
│ └── test_inventory.py # 单元测试
├── requirements.txt
├── docker-compose.yml # 本地环境编排
└── README.md关键点解析：分层架构：严格区分 models (ORM映射), schemas (API入参出参), services (业务逻辑), repositories (数据访问)。这样当数据库从Postgres换成MySQL时，你只需要改 repositories 层，services 层代码几乎不用动。
配置分离：config.py 使用 Pydantic Settings 读取环境变量，严禁在代码里硬编码密码或IP。
测试目录：没有测试的代码等于裸奔。tests 目录与 app 目录同级，便于 pytest 识别。核心代码实现
接下来是重头戏。我们将分步骤实现【二类疫苗】库存扣减的核心逻辑。
1. 数据模型定义
首先定义数据库模型。注意，这里我们特意增加了 version 字段，用于乐观锁。
# app/models/vaccine.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from app.models.order import Order # 假设Order模型存在Base = declarative_base()class Vaccine(Base):__tablename__ = 'vaccines'id = Column(Integer, primary_key=True, index=True)name = Column(String(100), nullable=False) # 疫苗名称，如23价肺炎疫苗stock = Column(Integer, nullable=False, default=0) # 当前库存price = Column(Integer, nullable=False) # 价格，单位：分valid_until = Column(DateTime, nullable=False) # 有效期version = Column(Integer, nullable=False, default=0) # 乐观锁版本号2. 库存扣减服务 (核心逻辑)
这是整个系统最复杂的部分。我们采用 “Redis预扣减 + DB最终一致” 的双层架构。
为什么这么设计？Redis层：抗压。绝大多数恶意刷单或高并发请求会被Redis挡掉，保护数据库。
DB层：准确。只有真正走到DB层的请求，才涉及真实的事务和锁。# app/services/inventory.py
import redis.asyncio as redis
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import update
from app.models.vaccine import Vaccine
from app.utils.logger import get_loggerlogger = get_logger(__name__)class InventoryService:def __init__(self, redis_client: redis.Redis, db: AsyncSession):self.redis = redis_clientself.db = dbasync def pre_deduct(self, vaccine_id: int, quantity: int = 1) - bool:第一步：Redis 预扣减利用 Lua 脚本保证原子性key = fvaccine:stock:{vaccine_id}# 注意：Lua脚本中的 KEYS[1] 是 key, ARGV[1] 是数量lua_script = local stock = tonumber(redis.call('get', KEYS[1]) or 0)if stock = tonumber(ARGV[1]) thenredis.call('decrby', KEYS[1), ARGV[1])return 1elsereturn 0end# 执行原子操作result = await self.redis.eval(lua_script, 1, key, quantity)return result == 1async def confirm_deduction(self, vaccine_id: int, order_id: str) - bool:第二步：DB 正式扣减 (乐观锁)# 1. 查询当前版本和库存stmt = (update(Vaccine).where(Vaccine.id == vaccine_id).where(Vaccine.version == self._get_current_version(vaccine_id)) # 需先查版本，或结合下文事务.values(stock=Vaccine.stock - 1, version=Vaccine.version + 1))# 为了演示清晰，这里简化处理，实际应使用 SELECT FOR UPDATE 或 先查后更# 严谨做法：在事务内先 SELECT ... FOR UPDATEtry:async with self.db.begin():# 锁行result = await self.db.execute(select(Vaccine).where(Vaccine.id == vaccine_id).with_for_update())vaccine = result.scalar_one_or_none()if not vaccine or vaccine.stock 1:raise Exception(库存不足或记录不存在)# 更新库存vaccine.stock -= 1vaccine.version += 1await self.db.commit()return Trueexcept Exception as e:await self.db.rollback()logger.error(fDB deduction failed for order {order_id}: {e})# 补偿：如果DB扣减失败，需要回滚Redisawait self.rollback_redis(vaccine_id)return Falseasync def rollback_redis(self, vaccine_id: int):补偿机制：DB失败时，Redis加回库存key = fvaccine:stock:{vaccine_id}await self.redis.incrby(key, 1)逐行避坑讲解：Lua脚本原子性：在Redis中，GET 和 DECRBY 分两步执行是非原子的。如果两个请求同时执行，都读到1，都执行减1，就会出错。Lua脚本在Redis单线程内执行，保证了原子性。
with_for_update()：这是SQLAlchemy提供的行锁机制。在高并发下，如果不加锁，两个事务可能同时读取到相同的 version，导致其中一个更新失效（乐观锁失效）。这里我们结合了悲观锁（行锁）和乐观锁（version）的思想，确保数据绝对安全。
补偿机制 rollback_redis：这是分布式事务的难点。如果Redis扣成功了，但DB因为网络超时失败了，如果不把Redis加回去，库存就“消失”了。这个回调函数必须在异常捕获中调用。3. 订单创建流程
将上述逻辑串联起来。
# app/services/order.py
from app.services.inventory import InventoryService
from app.models.order import Order
from uuid import uuid4async def create_order(db: AsyncSession, redis_client: redis.Redis, user_id: int, vaccine_id: int):inv_service = InventoryService(redis_client, db)# 1. 幂等性检查：防止同一用户同一疫苗重复预约existing_order = await db.execute(select(Order).where(Order.user_id == user_id, Order.vaccine_id == vaccine_id,Order.status == 'PENDING'))if existing_order.scalar_one_or_none():raise ValueError(请勿重复预约)# 2. Redis 预扣减success = await inv_service.pre_deduct(vaccine_id)if not success:raise ValueError(库存不足)try:# 3. 创建订单 (状态为 PENDING)order = Order(id=str(uuid4()),user_id=user_id,vaccine_id=vaccine_id,status='PENDING')db.add(order)await db.flush() # 立即生成ID，但不提交事务# 4. DB 正式扣减db_success = await inv_service.confirm_deduction(vaccine_id, order.id)if not db_success:raise Exception(DB扣减失败)# 5. 提交事务await db.commit()return orderexcept Exception as e:# 任何异常，都要回滚Redisawait inv_service.rollback_redis(vaccine_id)await db.rollback()raise e运行与测试
代码写完不能只看，必须跑。这里展示如何用 Docker Compose 一键启动环境。
docker-compose.yml 关键片段：
services:db:image: postgres:15environment:POSTGRES_DB: vaccine_dbPOSTGRES_USER: adminPOSTGRES_PASSWORD: admin123ports:- 5432:5432redis:image: redis:7-alpineports:- 6379:6379api:build: .ports:- 8000:8000depends_on:- db- redisenvironment:- DATABASE_URL=postgresql+asyncpg://admin:admin123@db:5432/vaccine_db- REDIS_URL=redis://redis:6379/0压测建议：
使用 locust 或 wrk 进行并发测试。场景：100个用户，同时抢购10个库存。
预期结果：Redis 中 vaccine:stock:1 从 10 变为 0。
DB 中 vaccines.stock 从 10 变为 0。
生成的订单数恰好为 10 条。
没有任何超卖（订单数 10）或少卖（订单数 10 且库存 0）。常见报错排查：Connection refused：检查 docker-compose 服务是否启动，端口映射是否正确。
Deadlock detected：PostgreSQL 锁冲突。检查是否在一个事务中更新了多行，且顺序不一致。解决方案：固定更新顺序。
Redis 连接超时：检查 redis.asyncio 是否配置了连接池，避免每次请求都建立新连接。优化扩展与进阶技巧
基础功能跑通后，我们需要考虑生产环境的复杂性。
1. 引入消息队列处理异步任务
支付回调、短信通知、库存释放，这些操作耗时较长，不应阻塞主流程。方案：引入 RabbitMQ 或 Kafka。
流程：用户支付成功 - 发送消息到 MQ。
消费者监听 MQ - 执行库存最终确认、发送短信。
如果消费者失败，进入死信队列，人工介入或重试。2. 数据一致性保障
在极端情况下（如 Redis 宕机），Redis 中的数据可能丢失。方案：持久化：Redis 开启 AOF 持久化。
对账任务：每天凌晨跑一个脚本，对比 Redis 中的库存和 DB 中的库存。如果有差异，以 DB 为准，修正 Redis。
降级策略：如果 Redis 不可用，直接走 DB 慢路径（加锁），虽然性能下降，但保证业务可用。3. 安全加固SQL注入：永远使用 ORM 或参数化查询，严禁拼接 SQL 字符串。
越权访问：在 Service 层严格校验 user_id 与 order_id 的归属关系。不能只靠前端传参。
限流：使用 Redis 令牌桶算法，对单个 IP 或用户进行 API 限流，防止恶意刷接口。4. 监控与告警Prometheus + Grafana：监控 QPS、响应时间、Redis 内存使用率、DB 连接池使用率。
日志追踪：使用 OpenTelemetry 给每个请求生成 TraceID，贯穿 API、Service、Repository 层，方便排查问题。一个真实的 Stack Overflow 案例：
很多开发者在 Stack Overflow 上提问：“为什么我的 Redis 扣减成功了，但 DB 没变？”
答案通常是：await db.commit() 没有被调用，或者异常捕获块中直接 return 了，导致事务未提交。
教训：在异步编程中，async/await 的错误处理比同步代码更隐蔽。一定要在 finally 块或 except 块中确保资源释放和状态回滚。
小结
通过这个【二类疫苗】管理系统的实战拆解，我们不仅仅写了一个 Demo，更构建了一套应对高并发、保证数据一致性的思维模型。
回顾核心知识点：分层架构：让代码可维护、可测试。
Redis + DB 双层扣减：兼顾性能与准确性。
乐观锁/悲观锁：解决并发冲突的核心手段。
补偿机制：分布式系统中，最终一致性的兜底方案。
幂等性设计：防止重复操作导致的数据错误。对于新手来说，不要指望一次就写出完美的代码。重要的是理解为什么要这么设计。当你下次面试被问到“如何处理高并发下的库存超卖”时，你能自信地画出架构图，说出 Redis Lua 脚本、DB 行锁、补偿事务这些关键词，并且能举出这个疫苗系统的例子，你就已经超过了 80% 的竞争者。
技术没有银弹，但好的架构能让你睡得安稳。
还有什么不懂的?评论区留言挨个回