二类疫苗管理系统从零搭建:新手避坑指南与实战拆解

面试被问原理答不上来，代码写出来却跑不通，这是很多刚入行或者转行做后端的新手最头疼的事。很多人以为写个增删改查就是开发，真到了实战项目里，才发现权限控制、数据一致性、并发安全全是坑。今天咱们不整虚的，直接上手做一个【二类疫苗】预约与库存管理系统的核心模块。

这不是一个玩具代码，而是一个能跑通业务逻辑、具备基本高可用特性的实战案例。我会把代码拆碎了揉烂了讲，专门针对【新手避坑】场景，告诉你为什么这么写，哪里容易炸，以及生产环境里该怎么兜底。

项目目标与核心痛点

做【二类疫苗】管理系统，表面看是管理药品库存，实则是在处理“高并发下的资源竞争”和“复杂状态机流转”。

为什么选这个场景？

业务闭环短：包含预约、锁定库存、支付（模拟）、核销、退款五个核心状态，逻辑清晰。
痛点集中：疫苗有有效期，库存有限，用户可能重复点击。这些都是面试高频考点。
技术覆盖面广：涉及数据库事务、乐观锁、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 = f"vaccine: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(f"DB 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 = f"vaccine: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:5432"redis:image: redis:7-alpineports:- "6379:6379"api:build: .ports:- "8000:8000"depends_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。
流程：
1. 用户支付成功 -> 发送消息到 MQ。
2. 消费者监听 MQ -> 执行库存最终确认、发送短信。
3. 如果消费者失败，进入死信队列，人工介入或重试。

2. 数据一致性保障

在极端情况下（如 Redis 宕机），Redis 中的数据可能丢失。

方案：
1. 持久化：Redis 开启 AOF 持久化。
2. 对账任务：每天凌晨跑一个脚本，对比 Redis 中的库存和 DB 中的库存。如果有差异，以 DB 为准，修正 Redis。
3. 降级策略：如果 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 块中确保资源释放和状态回滚。