C店实战：从零搭建高可用电商后端完整示例

面试被问原理答不上来？这不仅是技术短板，更是工程思维的缺失。今天用C店这个极简但完整的电商后端案例，带你彻底搞懂高并发下的核心逻辑。

我们不再满足于“跑通代码”，而是聚焦完整示例的实战落地。很多初学者写Demo时，数据库一崩、订单一重、库存一超卖，立马就懵了。这背后是缺乏对分布式场景下一致性、幂等性的深入理解。

本文将基于Python + FastAPI + Redis + MySQL，从零搭建一个具备生产级思维的C店系统。不堆砌花哨功能，只打磨核心链路：商品浏览、库存扣减、订单创建、支付回调。每一步都拆解原理，每一段代码都标注关键决策点。

项目目标：定义什么是“可用”的C店

在写第一行代码前，先明确目标。C店不是淘宝，不需要秒杀百万级QPS，但必须具备以下工程特性：

数据一致性：库存不能超卖，订单状态必须准确。
幂等性保障：网络抖动导致重复请求时，不能产生重复订单或重复扣款。
高可用设计：单点故障不导致整个服务崩溃，关键依赖（如Redis）有降级策略。
可观测性：日志结构化，关键链路可追踪，错误可定位。

这里参考了Stack Overflow上关于“分布式库存扣减方案”的高赞讨论，核心共识是：本地消息表 + 异步补偿 或 Redis预扣减 + DB最终一致 是中小规模电商最稳妥的选择。我们采用后者，因为性能更好，且C店场景下对极端一致性的容忍度略高。

目录结构：模块化思维落地

好的代码结构是维护性的基石。我们采用分层架构，职责清晰：

c_store/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI入口，挂载路由与中间件
│   ├── core/
│   │   ├── config.py        # 配置管理（Pydantic Settings）
│   │   ├── security.py      # JWT鉴权逻辑
│   │   └── exceptions.py    # 自定义异常处理
│   ├── models/
│   │   ├── user.py          # 用户模型
│   │   ├── product.py       # 商品模型
│   │   └── order.py         # 订单模型
│   ├── schemas/
│   │   ├── product.py       # Pydantic Schema，用于请求/响应验证
│   │   └── order.py
│   ├── services/
│   │   ├── product_service.py  # 商品业务逻辑
│   │   ├── inventory_service.py # 库存核心逻辑（Redis+DB）
│   │   └── order_service.py    # 订单创建与状态流转
│   ├── repositories/
│   │   ├── base.py          # 通用DB操作基类
│   │   ├── product_repo.py
│   │   └── order_repo.py
│   └── api/
│       ├── v1/
│       │   ├── products.py  # /api/v1/products 路由
│       │   ├── orders.py    # /api/v1/orders 路由
│       │   └── health.py    # 健康检查
│       └── deps.py          # 依赖注入（DB Session, Current User）
├── alembic/                 # 数据库迁移脚本
├── tests/
│   ├── conftest.py
│   └── test_order_flow.py
├── requirements.txt
├── docker-compose.yml       # 一键启动MySQL/Redis/FastAPI
└── README.md

关键设计说明：

Services层 不直接操作DB，而是调用Repositories，便于单元测试时Mock DB。
Schemas 与 Models 分离，避免ORM模型直接暴露给API，提升安全性与灵活性。
Alembic 管理数据库版本，杜绝手动改表结构，保证团队协作时的环境一致性。

核心代码实现：库存扣减与订单创建

这是整个C店最核心的部分。我们将重点讲解Redis预扣减 + DB最终一致 的实现细节。

1. 库存服务：原子性操作保障

# app/services/inventory_service.py
import redis.asyncio as redis
from sqlalchemy.ext.asyncio import AsyncSession
from app.repositories.product_repo import ProductRepository
from app.core.exceptions import StockInsufficientErrorclass InventoryService:def __init__(self, redis_client: redis.Redis, db: AsyncSession):self.redis = redis_clientself.product_repo = ProductRepository(db)async def pre_deduct_stock(self, product_id: int, quantity: int) -> bool:"""预扣减库存：在Redis中执行原子性扣减返回True表示扣减成功，False表示库存不足"""key = f"stock:{product_id}"# 使用Lua脚本保证读取和扣减的原子性，避免竞态条件lua_script = """local stock = tonumber(redis.call('GET', KEYS[1]))if stock == false thenreturn -1endif stock >= tonumber(ARGV[1]) thenreturn redis.call('DECRBY', KEYS[1], ARGV[1])elsereturn -1end"""result = await self.redis.eval(lua_script, 1, key, quantity)if result == -1:# 库存不足或Key不存在if await self.redis.exists(key) == 0:# Key不存在，可能是未预热，回源DB查询并设置await self._load_stock_to_redis(product_id)# 重新尝试一次result = await self.redis.eval(lua_script, 1, key, quantity)if result == -1:return Falseelse:return Falsereturn Trueasync def confirm_deduct_stock(self, product_id: int, quantity: int):"""确认扣减：订单支付成功后，同步DB库存注意：这里不操作Redis，Redis数据仅作为缓存与预扣减依据"""# 使用SELECT ... FOR UPDATE 行锁，防止并发更新product = await self.product_repo.get_for_update(product_id)if not product or product.stock < quantity:# 理论上不应该发生，因为预扣减已拦截raise StockInsufficientError(f"DB stock insufficient for {product_id}")product.stock -= quantityawait self.db.commit()async def rollback_stock(self, product_id: int, quantity: int):"""回滚库存：订单取消或支付超时后，恢复Redis与DB库存"""key = f"stock:{product_id}"await self.redis.incrby(key, quantity)# 注意：DB库存的回滚通常由异步任务处理，避免阻塞主流程# 这里简化处理，实际项目中应发送MQ消息触发DB回滚

逐行讲解：

Lua脚本 是Redis保证原子性的关键。直接写 GET 再 DECRBY 是非原子的，高并发下会超卖。
Key不存在处理：冷启动时Redis无数据，需回源DB加载。这里做了重试机制，避免首次请求失败。
确认扣减：支付成功后才真正修改DB。使用 FOR UPDATE 行锁是MySQL并发控制的经典手段，防止两个请求同时读到相同库存值。

2. 订单服务：幂等性与状态机

# app/services/order_service.py
import uuid
from datetime import datetime, timedelta
from app.services.inventory_service import InventoryService
from app.repositories.order_repo import OrderRepository
from app.models.order import Order, OrderStatus
from sqlalchemy.ext.asyncio import AsyncSessionclass OrderService:def __init__(self, db: AsyncSession, inventory_service: InventoryService):self.db = dbself.inventory_service = inventory_serviceself.order_repo = OrderRepository(db)async def create_order(self, user_id: int, product_id: int, quantity: int, idempotency_key: str) -> Order:"""创建订单，核心在于幂等性处理"""# 1. 幂等性检查：根据idempotency_key查询是否已存在订单existing_order = await self.order_repo.find_by_idempotency_key(idempotency_key)if existing_order:# 直接返回已存在的订单，不重复创建return existing_order# 2. 预扣减库存stock_ok = await self.inventory_service.pre_deduct_stock(product_id, quantity)if not stock_ok:raise StockInsufficientError("Insufficient stock")# 3. 创建订单记录，状态为"待支付"order = Order(order_no=f"ORD{uuid.uuid4().hex[:12]}",user_id=user_id,product_id=product_id,quantity=quantity,status=OrderStatus.PENDING_PAYMENT,idempotency_key=idempotency_key,expire_at=datetime.now() + timedelta(minutes=30)  # 30分钟未支付自动取消)self.db.add(order)await self.db.flush()  # 获取order.id# 4. 发送延迟消息（简化示例，实际使用RabbitMQ延迟队列或Redis ZSET）# await self.message_queue.send_delayed(order.order_no, delay=1800)await self.db.commit()return order

关键细节：

Idempotency Key：由前端生成唯一UUID，贯穿整个请求。这是防止重复下单的最可靠方式，比“查询+插入”更健壮。
状态机：订单状态必须严格流转，禁止直接从“待支付”跳到“已完成”。后续可引入状态机库（如python-statemachine）简化逻辑。

运行与测试：验证工程化能力

代码写得再好，跑不起来等于零。我们用 docker-compose 一键启动环境。

# docker-compose.yml
version: '3.8'
services:mysql:image: mysql:8.0environment:MYSQL_ROOT_PASSWORD: rootMYSQL_DATABASE: c_storeports:- "3306:3306"volumes:- mysql_data:/var/lib/mysqlredis:image: redis:7-alpineports:- "6379:6379"app:build: .ports:- "8000:8000"environment:DATABASE_URL: "mysql+aiomysql://root:root@mysql:3306/c_store"REDIS_URL: "redis://redis:6379/0"depends_on:- mysql- redisvolumes:- .:/appvolumes:mysql_data:

测试重点：

并发压测：用 locust 模拟100个用户同时抢购10件库存商品，验证无超卖。
幂等性测试：相同 idempotency_key 连续请求10次，DB中只产生1条订单记录。
故障注入：手动停止Redis，观察系统是否降级（如只读DB库存，或快速失败返回友好提示），而非抛出500错误。

优化扩展：从Demo到生产

C店基础版跑通后，还有几个方向值得深入：

数据库分库分表：当订单量突破千万级，单表性能瓶颈明显。可按 user_id 哈希分表，使用 ShardingSphere 或 Vitess 中间件。
缓存预热：服务启动时，主动将热门商品库存加载到Redis，避免冷启动时的DB压力。
监控告警：集成 Prometheus + Grafana，监控订单创建成功率、Redis内存使用率、DB连接池状态。
日志追踪：使用 OpenTelemetry 实现全链路追踪，将 request_id 贯穿HTTP请求、DB查询、Redis操作，便于问题排查。