第30章：【中级篇综合实战】FastAPI 构建电商订单中台 API

1. 项目背景

业务场景

"极速商城"是一个快速增长的电商平台，日均订单量从 500 涨到 5 万，原有单体架构已经不堪重负。CTO 决定搭建订单中台——一个承上启下的核心服务层，对上为前端/App/BFF 提供统一的订单能力，对下协调库存、支付、物流等基础服务。

这是中级篇（第 16-29 章）的综合实战。你将独立交付一个接近生产标准的订单中台，它需要融合本级别学到的所有技术：

验收标准

• 核心链路（下单→扣库存→创建支付）P95 < 200ms（缓存命中时）
• 5xx 错误率 < 0.1%
• 1000 并发压测通过
• 关键告警可触发（5xx 错误率 > 1%、P99 > 1s、支付成功率 < 95%）
• OpenAPI 文档完整，支持商户专用视图

2. 项目设计

场景：项目 Kick-off 会议。大师在白板上画出了订单中台的整体架构。

小胖：（看着白板上密密麻麻的箭头）“这架构图比基础篇的任务系统复杂 10 倍——这么多服务互相调来调去，真的有必要吗？”

大师：“不是’这次更复杂’，而是’这次更真实’。真实生产的订单中台就是要面对商品、购物车、下单、库存、支付、物流、通知——7 个域。但我们不追求全功能——抓核心路径：下单→扣库存→支付。这条链路通了，中台就立住了。”

小白：“那这次怎么组织代码？还是 domains/ 方式吗？”

大师：“对，但这次中台本身要拆成两个独立的 FastAPI 应用——order-service（订单服务）和 inventory-service（库存服务）。两者通过第 28 章的 ServiceClient + 熔断器通信。每个服务有自己的数据库，独立部署，独立扩缩容。”

小胖：“数据库也要拆？不能共享一个 DB 吗？”

大师：“共享数据库是微服务的’反模式’——表面上拆了服务，底下共享一个数据库，最终还是耦合在一起。库存服务改表结构可能影响订单服务的查询——这就是’分布式单体’。正确的做法：每个微服务有自己的数据库，互相之间只通过 API 通信。”

技术映射：数据库隔离（Database per Service）是微服务架构的核心原则。它保证了服务之间的低耦合——订单服务不需要知道库存数据的存储结构，也不需要知道库存服务用的是 PostgreSQL 还是 MongoDB。

小白：“那这次验收标准能达成吗？P95 < 200ms 看起来很难。”

大师："分阶段来：

任何性能目标都是迭代出来的，不是一开始就能达成的。"

小胖：“那我们第一个迭代具体做哪几件事？我有点迷失在这么多技术栈里——不知道该从哪下手。”

大师：“按’核心链路优先’原则——下单→扣库存→支付 这条链路先跑通，再逐步加缓存、异步、观测。第一个迭代只做三件事：① 两个服务的脚手架 + 数据库模型 ② 同步下单接口（暂不接缓存和消息队列）③ 库存扣减接口。这三个做完就能端到端验证——虽然 P95 可能 800ms，但至少链路是通的。”

小白：“Outbox 模式一定要在第一版就做吗？我觉得可以先用手动 try-except-rollback，等出问题了再加 Outbox。”

大师：（摇头）“不。Outbox 是可靠性基础——没有它，订单创建成功但库存扣减消息丢失，用户付了钱却没扣库存，这是线上事故。可靠性基建不能在’出问题之后’才补——就像盖房子不能先盖 3 层再加地基。Outbox 的实现成本很低（一张表 + Worker 轮询），但缺了它，后续所有异步流程都没有可靠性保障。”

技术映射：Outbox Pattern 的核心价值是保证"业务操作"与"消息发送"的原子性——它们在同一个数据库事务中，要么都成功，要么都失败。这是微服务间可靠通信的基石——没有它，分布式事务（Saga 补偿）根本无法实现。

小胖：“我还有一个担心——测试怎么办？两个独立的服务，数据库也独立——测试环境怎么搭？”

大师："好问题。测试分三层：

3. 项目实战——分 6 步交付订单中台

环境准备

# 快速启动所有基础设施
docker compose -f docker-compose.infra.yml up -d
# 包含: PostgreSQL×2, Redis, RabbitMQ, Prometheus, Grafana, Jaeger

pip install fastapi==0.115.6 uvicorn==0.34.0 sqlalchemy==2.0.36 \\
asyncpg==0.30.0 redis==5.2.0 celery==5.4.0 httpx==0.27.0 \\
prometheus-fastapi-instrumentator==7.0.0 python-jose[cryptography]==3.3.0 \\
passlib[bcrypt]==1.7.4 k6 # 或操作系统安装

分步实现

步骤一：项目结构与双服务部署（目标：搭建能跑通的骨架）

坑预警：两个服务用同一个 app/core/ 共享核心模块时，如果一方修改了核心模块的 dependencies.py（如增加一个新的 Depends），另一方可能因为没有更新 import 而在运行时抛出 AttributeError。解决：用 monorepo + Git submodule 管理共享核心，或在 CI 中确保两个服务始终使用同一版本的共享核心。

order-platform/
├── services/
│ ├── order-service/ # 订单服务（主要业务逻辑）
│ │ ├── app/
│ │ │ ├── main.py # FastAPI 应用（端口 8001）
│ │ │ ├── bootstrap.py
│ │ │ ├── core/ # 配置、数据库、安全、依赖
│ │ │ ├── models/ # Order, OutboxMessage
│ │ │ ├── schemas/ # Pydantic Schema
│ │ │ ├── domains/order/ # api, service, repository
│ │ │ └── infrastructure/ # cache, service_client, cel_tasks
│ │ ├── Dockerfile
│ │ └── alembic/
│ │
│ └── inventory-service/ # 库存服务（独立数据库）
│ ├── app/
│ │ ├── main.py # FastAPI 应用（端口 8002）
│ │ ├── models/ # Inventory, InventoryLog
│ │ └── domains/inventory/ # api, service, repository
│ └── Dockerfile
│
├── docker-compose.yml # 完整编排（4+ 容器）
├── docker-compose.infra.yml # 仅基础设施
├── k8s/ # K8s 部署文件
├── scripts/loadtest/ # k6 压测脚本
└── config/ # Prometheus, Grafana dashboards

步骤二：订单服务核心实现（融合第 16-25 章）

订单创建主流程（services/order-service/app/domains/order/service.py）：

class OrderService:
"""订单服务——融合多章技术"""

async def create_order(
self, db: AsyncSession, user_id: int, data: dict,
bg: BackgroundTasks, cache: RedisCache,
) –> Order:
# ── Step 1: 缓存查询商品信息（第 19 章）──
cache_key = f"product:{data['product_id']}"
product = await cache.get(cache_key)
if not product:
# 缓存未命中 → 调商品服务（熔断保护 第 28 章）
product = await product_client.request("GET", f"/products/{data['product_id']}")
await cache.set(cache_key, product, ttl=3600)

# ── Step 2: 同步创建订单 + Outbox 写入（第 28 章）──
async with db.begin():
order = Order(
user_id=user_id,
product_name=product["name"],
quantity=data["quantity"],
unit_price=product["price"],
total_amount=data["quantity"] * product["price"],
status="pending",
)
db.add(order)
await db.flush()

# Outbox: 库存扣减消息
db.add(OutboxMessage(
event_type="inventory_deduct",
payload=json.dumps({"order_id": order.id, "product_id": data["product_id"], "quantity": data["quantity"]}),
))

# ── Step 3: 异步通知（Celery 第 20 章）──
bg.add_task(send_order_sms.delay, order.id, user_id)

# ── Step 4: 自定义 Prometheus 指标（第 24 章）──
ORDER_COUNTER.labels(status="created").inc()

return order

async def confirm_order(self, db: AsyncSession, order_id: int, user_id: int):
"""确认订单——含权限校验（第 22 章）"""
order = await db.get(Order, order_id)
if not order:
raise NotFoundException("订单不存在")
if order.user_id != user_id:
raise ForbiddenException("无权操作此订单") # 对象级权限

order.status = "confirmed"
await db.commit()

# WebSocket 推送状态变更（第 29 章）
await ws_backplane.publish("order_status", {
"order_id": order_id, "status": "confirmed"
})
return order

步骤三：库存服务独立实现

services/inventory-service/app/domains/inventory/api.py：

router = APIRouter(prefix="/inventory", tags=["库存管理"])

@router.post("/deduct", summary="扣减库存")
async def deduct_inventory(req: DeductRequest, db: AsyncSession = Depends(get_db)):
"""
扣减库存——供订单服务的 Saga 调用
幂等保护：同一 order_id 的扣减只执行一次
"""
# 幂等性检查
existing = await db.execute(
select(InventoryLog).where(InventoryLog.order_id == req.order_id)
)
if existing.scalar_one_or_none():
return {"status": "already_deducted"}

async with db.begin():
# 行级锁（SELECT … FOR UPDATE）防止超卖
item = (await db.execute(
select(Inventory).where(Inventory.product_id == req.product_id)
.with_for_update()
)).scalar_one_or_none()

if not item or item.stock < req.quantity:
raise BusinessException(code=2001, message="库存不足")

item.stock -= req.quantity

# 记录操作日志（幂等键：order_id）
db.add(InventoryLog(order_id=req.order_id, product_id=req.product_id, quantity=–req.quantity))

return {"status": "deducted", "remaining_stock": item.stock}

坑预警：with_for_update() 必须放在 async with db.begin() 事务块内——如果放在事务外（自动提交模式），行级锁在查询完立刻释放，并发场景下形同虚设。验证方法：用两个并发请求（k6 或 pytest-asyncio 并发）同时扣减最后 1 件库存——正确结果是一个返回成功、一个返回 库存不足。

步骤四：Docker Compose 编排（第 26 章）

services:
order-api:
build: ./services/order–service
ports: ["8001:8001"]
environment:
DATABASE_URL: postgresql+asyncpg://user:pass@order–pg:5432/orderdb
REDIS_URL: redis://redis:6379/0
INVENTORY_SERVICE_URL: http://inventory–api:8002/api/v1

inventory-api:
build: ./services/inventory–service
ports: ["8002:8002"]
environment:
DATABASE_URL: postgresql+asyncpg://user:pass@inventory–pg:5432/inventorydb

order-pg:
image: postgres:16–alpine
environment:
POSTGRES_DATABASE: orderdb
inventory-pg:
image: postgres:16–alpine
environment:
POSTGRES_DATABASE: inventorydb

redis:
image: redis:7–alpine
celery-worker:
build: ./services/order–service
command: celery –A app.infrastructure.celery_app worker –Q notifications ––concurrency=2

坑预警：depends_on 只保证容器启动顺序，不保证 PostgreSQL 内部的数据库已经 ready 接受连接。如果 order-api 在 postgres 的 pg_isready 返回 true 之前就尝试连接，会报 Connection refused。解决：在 depends_on 中加 condition: service_healthy（需在 postgres 服务中定义 healthcheck）。对于生产部署，K8s 用 initContainer 等待数据库就绪。

步骤五：k6 压测脚本（第 25 章）

scripts/loadtest/order_platform.js：

import http from 'k6/http';
import { check, sleep } from 'k6';
import { Rate, Trend } from 'k6/metrics';

const errorRate = new Rate('errors');
const orderDuration = new Trend('order_duration');

export const options = {
stages: [
{ duration: '2m', target: 100 },
{ duration: '3m', target: 100 },
{ duration: '2m', target: 500 },
{ duration: '3m', target: 500 },
{ duration: '2m', target: 1000 },
{ duration: '3m', target: 1000 },
{ duration: '2m', target: 0 },
],
thresholds: {
'http_req_duration': ['p(95)<500'],
'http_req_failed': ['rate<0.01'],
'order_duration': ['p(95)<800'],
},
};

export default function () {
// 注册+登录流程（复用第 25 章脚本逻辑）
const orderStart = Date.now();
const resp = http.post(`${BASE_URL}/orders`, JSON.stringify({
product_id: 1, quantity: 1,
}), { headers: authHeader });
orderDuration.add(Date.now() – orderStart);
errorRate.add(resp.status !== 201);
sleep(0.5);
}

k6 run scripts/loadtest/order_platform.js

步骤六：可观测性与告警配置（第 24 章）

# config/prometheus_alerts.yml
groups:
– name: order_platform
rules:
– alert: OrderCreationErrorRateHigh
expr: rate(http_requests_total{handler="/api/v1/orders",status=~"5.."}[5m]) > 0.01
for: 5m
annotations:
summary: "订单创建 5xx 错误率 > 1%"

– alert: OrderP99High
expr: histogram_quantile(0.99, http_request_duration_seconds_bucket{handler="/api/v1/orders"}[5m]) > 1.0
for: 5m
annotations:
summary: "订单创建 P99 > 1s"

– alert: PaymentSuccessRateLow
expr: rate(payment_success_total[10m]) / rate(payment_total[10m]) < 0.95
for: 10m
annotations:
summary: "支付成功率 < 95%"

完整代码清单

本项目完整代码见 column/code/chapter30/，包含两个完整服务和所有基础设施配置。

测试验证

下面提供一套完整的验证步骤。建议在 CI 中把这些命令写成自动化脚本（scripts/smoke_test.sh）。

# ═══════════════════════════════════════════════
# 1. 环境启动与健康检查
# ═══════════════════════════════════════════════
docker compose up -d

# 等待所有服务健康
for i in {1..30}; do
if curl -s http://localhost:8001/api/v1/health && \\
curl -s http://localhost:8002/api/v1/health; then
echo "✓ 所有服务就绪"
break
fi
sleep 2
done

# ═══════════════════════════════════════════════
# 2. 正常下单全链路验证
# ═══════════════════════════════════════════════

# 2a. 注册 + 登录
TOKEN=$(curl -s -X POST http://localhost:8001/api/v1/auth/login \\
-H "Content-Type: application/x-www-form-urlencoded" \\
-d "username=testuser&password=Test123456" | python -c "import sys,json; print(json.load(sys.stdin)['access_token'])")

# 2b. 创建订单
curl -s -X POST http://localhost:8001/api/v1/orders \\
-H "Authorization: Bearer $TOKEN" \\
-H "Content-Type: application/json" \\
-d '{"product_id":1,"quantity":2}' | python -m json.tool
# 预期: {"code":0,"data":{"id":1,"status":"pending",…}}

# 2c. 验证库存已扣减
curl -s http://localhost:8002/api/v1/inventory/1 | python -m json.tool
# 预期: stock 从 10 变为 8（扣了 2）

# ═══════════════════════════════════════════════
# 3. 异常场景验证
# ═══════════════════════════════════════════════

# 3a. 库存不足
curl -s -X POST http://localhost:8001/api/v1/orders \\
-H "Authorization: Bearer $TOKEN" \\
-H "Content-Type: application/json" \\
-d '{"product_id":1,"quantity":999}'
# 预期: {"code":2001,"message":"库存不足"}

# 3b. 未登录访问（验证 JWT 中间件生效）
curl -s http://localhost:8001/api/v1/orders -w "\\nHTTP %{http_code}"
# 预期: HTTP 401

# 3c. 访问不存在的订单
curl -s http://localhost:8001/api/v1/orders/99999 \\
-H "Authorization: Bearer $TOKEN" -w "\\nHTTP %{http_code}"
# 预期: HTTP 404

# 3d. 权限校验——用户只能看自己的订单（第 22 章复习）
# 用另一个用户的 Token 访问——应返回 403 或空列表
ANOTHER_TOKEN=$(curl -s -X POST http://localhost:8001/api/v1/auth/login \\
-H "Content-Type: application/x-www-form-urlencoded" \\
-d "username=anotheruser&password=Test123456" | python -c "import sys,json; print(json.load(sys.stdin)['access_token'])")
curl -s http://localhost:8001/api/v1/orders/1 \\
-H "Authorization: Bearer $ANOTHER_TOKEN" -w "\\nHTTP %{http_code}"
# 预期: HTTP 403

# ═══════════════════════════════════════════════
# 4. 熔断器验证（需要手动触发）
# ═══════════════════════════════════════════════
# 停止库存服务
docker compose stop inventory-api
# 再次创建订单——熔断器应在 5 次失败后 OPEN
for i in {1..6}; do
curl -s -X POST http://localhost:8001/api/v1/orders \\
-H "Authorization: Bearer $TOKEN" \\
-H "Content-Type: application/json" \\
-d '{"product_id":1,"quantity":1}' \\
-w " → HTTP %{http_code}\\n"
done
# 预期: 前 5 次返回 503（重试中），第 6 次直接拒绝（熔断器 OPEN）
docker compose start inventory-api # 恢复库存服务

# ═══════════════════════════════════════════════
# 5. 压测验证（第 25 章）
# ═══════════════════════════════════════════════
k6 run scripts/loadtest/order_platform.js

# ═══════════════════════════════════════════════
# 6. 观测面板验证（第 24 章）
# ═══════════════════════════════════════════════
# Grafana: http://localhost:3000 (admin/admin)
# Jaeger: http://localhost:16686
# 搜索 trace → 应看到: order-api.create_order → inventory-api.deduct → DB query

# ═══════════════════════════════════════════════
# 7. 验收检查清单
# ═══════════════════════════════════════════════
# ☐ 1000 并发压测通过
# ☐ P95 延迟 < 500ms (含缓存命中)
# ☐ 5xx 错误率 < 0.1%
# ☐ 库存超卖测试通过（并发扣减最后 1 件库存——只有一个成功）
# ☐ 熔断器在 5 次失败后自动 OPEN
# ☐ Grafana 面板可查看实时 QPS/P95/错误率
# ☐ Jaeger 可追踪完整链路（API → DB → Redis → Inventory Service）
# ☐ Prometheus 告警规则已配置

4. 项目总结

优点 & 缺点对比

适用场景

✓ 订单中台适用于：

✗ 不适用：

注意事项

常见踩坑经验

案例一：Saga 补偿中库存扣了又还导致少货

• 现象：Saga Step 2 扣库存成功，Step 3 创建支付失败——触发补偿释放库存。但用户又来一次下单——库存被"释放"了两次。
• 根因：补偿操作 release_inventory 没有幂等性——同一个 order_id 的释放被执行了两次（重试机制触发）。
• 解决：库存操作日志表加 order_id UNIQUE——第二次释放时直接跳过。

案例二：两个服务各自缓存导致数据不一致

• 现象：商品价格在商品服务中改为 199，但订单服务的 Redis 缓存中还是 99——用户下单用旧价格。
• 根因：订单服务缓存了商品信息，商品服务改了价格但没有通知订单服务淘汰缓存。
• 解决：商品服务在更新价格后，Publish 一条 Redis 消息 product:price_changed:{product_id}，订单服务 Subscribe 并淘汰对应的缓存。

案例三：依赖的库存服务挂了导致熔断器一直 OPEN

• 现象：库存服务故障修复后，订单服务仍然返回"服务不可用"——因为熔断器还停留在 OPEN 状态。
• 根因：熔断器的 HALF_OPEN 测试请求是在有真实请求时才触发的。如果故障期间没人下单，熔断器永远不会从 OPEN 变成 HALF_OPEN。
• 解决：添加独立的健康检查探测——熔断器自己的后台线程定期发试探请求（不与用户请求耦合）。

案例四：两个服务的并发压测时 Celery Worker 成为瓶颈

• 现象：1000 并发压测通过，但观察 Celery Worker 日志——通知任务排队超过 5000 条，延迟长达 3 分钟。用户下单 3 分钟后才收到短信。
• 根因：压测脚本中每 0.5 秒下一次单，每个下单触发一个通知任务——1000 VU × 2 单/秒 = 2000 条通知/秒。但 Celery Worker 只有 2 个并发（–concurrency=2），处理一条通知需要 200ms——实际处理能力 = 2/0.2 = 10 条/秒。2000 条/秒 vs 10 条/秒 = 200 倍的积压。
• 解决：通知任务的优先级低于核心下单链路——积压是预期的。关键是要监控队列深度（Flower Dashboard 或自定义 Prometheus 指标 celery_queue_length）。当队列深度超过 5000 时，增加 Worker 副本（K8s HPA），或暂时丢弃低优先级的通知（只发关键通知）。

案例五：订单和库存两个服务的 Alembic 迁移时间戳冲突

• 现象：两个服务都有 20260115_init.py 迁移文件——合并到 monorepo 后，alembic 版本链混乱。而且两个服务如果共享 PostgreSQL（分 Schema），各自迁移脚本可能修改对方的 Schema。
• 根因：两个服务独立生成迁移文件时用了同一天日期，没有做文件名去重。
• 解决：迁移文件名加服务名前缀——order_20260115_init.py、inventory_20260115_init.py。更根本的是——数据库隔离后，两个服务各有自己的数据库，各自维护迁移链，互不干扰。

思考题

答案提示：第 1 题购物车数据 cart:{user_id} 存 Redis Hash（key=product_id, value=quantity），提交时用 Redis Pipeline 原子读取全部商品并清空购物车。第 2 题：TCC 比 Saga 更强的一致性（预留资源不会被超卖两次），但每个资源接口需要提供 Try/Confirm/Cancel 三个方法——代码量是 Saga 的 3 倍。TCC 适合金融支付场景（预留余额），Saga 适合电商下单场景（库存扣减失败可以补偿）。第 3 题：第一个瓶颈可能是数据库连接池（5000 个并发 vs 20 个连接）——解决方案是连接池扩展 + 读写分离 + 热点库存数据上 Redis 做"预扣库存"减少数据库行锁竞争。第 39 章 SRE 落地深入容量规划。

中级篇完结。从第 16 章的生产级架构到第 30 章交付一个电商订单中台——你已经掌握了 FastAPI 在分布式场景下的工程化能力。第 31 章开始进入高级篇——源码剖析、极端性能优化、自定义扩展、企业级安全治理、SRE 落地。

延伸阅读与资源

NumPy 从入门到生产落地：全链路实战指南（科学计算/向量化） Redis 8 实战精讲：从 CRUD 到源码，构建高可用缓存系统 Redis 实战修炼与原理进阶 Python 3实战精进：从脚本到高并发订单引擎 python入门：Rquests从菜鸟脚本到企业级SDK的网络实战圣经 Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地 MongoDB 实战进阶与内核修炼 后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战 10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用 后端工程师转型AI第一课-Ollama 与私有化大模型实战 大型语言模型(LLM) vLLM 高性能推理落地实战 Agent开发之LlamaIndex 实战修炼与源码进阶 大语言模型Transformers 实战修炼与源码剖析