当前位置：首页 > news >正文

FastAPI分布式系统实战：拆解分布式系统中常见问题及解决方案

news 2026/5/12 10:01:47

“你是不是也遇到过——明明本地跑得好好的FastAPI服务，一放到分布式环境里，就像脱缰的野马，各种超时、数据对不上、资源争抢？” 🤔

📌 本文摘要：从实际踩坑案例出发，拆解分布式系统中服务调用超时、数据一致性、分布式锁三大高频问题。用生活化的比喻讲透原理，并给出可直接粘贴的FastAPI代码片段。帮你从“能跑”进化到“稳如狗”。

📋 今天要聊的坑

1️⃣ 服务间调用：超时、重试、幂等——打电话占线怎么办？

2️⃣ 分布式锁：抢钥匙的哲学——别让两个服务员同时收拾一张桌子

3️⃣ 数据最终一致性：先扣钱还是先发货？——用消息队列来解耦

🎯 为什么FastAPI在分布式里容易“翻车”？

FastAPI的异步特性让它像一台高性能跑车，但分布式系统不是单车道——它更像一个复杂的城市交通网。你的服务要和别人通信、共享资源、保证数据不打架。最常见的三大痛点：

🚦 网络不可靠：服务A调用服务B，可能超时、断连，甚至B处理完了但A没收到响应。

🔑 资源竞争：多个实例同时操作同一笔订单，搞出负数库存。

📦 数据一致性：跨库操作，要么都成功，要么都失败，但分布式事务太重了。

接下来咱们一个一个解决，建议收藏后反复看。

☎️ 问题一：服务调用像“打电话占线”

想象你给餐厅打电话订位（服务A），服务员接起来后去查记录（服务B处理中），但这时电话突然断了——你可能会重打，但服务员那边可能已经帮你预留了位置。这就是重复执行的隐患。

解决方案：超时控制 + 重试 + 幂等性

在FastAPI里，我习惯用httpx.AsyncClient，并且必须显式设置超时。千万别用默认值，否则生产环境一个慢请求就能拖垮整个链路。

# client.py
import httpxasync def call_service_b(order_id: str):async with httpx.AsyncClient(timeout=10.0) as client:  # 总超时10秒try:resp = await client.post("http://service-b/process",json={"order_id": order_id},headers={"Idempotency-Key": order_id}  # 幂等键)resp.raise_for_status()return resp.json()except httpx.TimeoutException:# 这里记录日志，触发补偿或告警，但不要直接重试！print(f"调用超时，但可能已处理，需查状态 order_id={order_id}")# 最佳实践：查询下游状态，决定是否重试

这里的关键是Idempotency-Key头，下游服务必须根据这个key保证同一个请求只处理一次。自己实现时可以用Redis记录已处理的key。

⚠️ 重点警告：重试一定要配合幂等，否则就是灾难。我见过因为重试导致重复扣款的案例，最后用唯一索引+业务状态机才解决。

🔐 问题二：抢钥匙——分布式锁

多个服务实例同时操作同一份数据，就像几个人抢同一间厕所，不加锁就会“撞车”。以最常见的库存扣减为例：

# 错误示范：不加锁
async def deduct_stock(product_id: str, quantity: int):product = await db.products.find_one({"_id": product_id})if product.stock >= quantity:product.stock -= quantityawait db.products.save(product)  # 两个并发可能同时读到stock=10，都减到9，实际只减了1

正确姿势：用Redis实现分布式锁（推荐Redlock算法，但简单场景用单点锁+过期时间即可）。

# 使用 aioredis 实现锁
import aioredis
import asyncioredis = aioredis.from_url("redis://localhost")async def acquire_lock(lock_key: str, timeout: int = 10):# SET NX 且设置过期时间，防止死锁locked = await redis.set(lock_key, "locked", nx=True, ex=timeout)return lockedasync def release_lock(lock_key: str):await redis.delete(lock_key)async def deduct_stock_with_lock(product_id: str, quantity: int):lock_key = f"lock:product:{product_id}"if not await acquire_lock(lock_key):raise Exception("系统繁忙，请稍后重试")try:# 业务逻辑product = await db.products.find_one({"_id": product_id})if product.stock >= quantity:product.stock -= quantityawait db.products.save(product)return Truereturn Falsefinally:await release_lock(lock_key)  # 一定要释放

这里有个坑：锁的过期时间要大于业务最大执行时间，否则业务没做完锁就自动释放了，别的请求又进来了。我的经验是设置30秒，并配合看门狗（Watch Dog）机制续期，但简单场景可以预估时间设长一点。

📬 问题三：数据一致性——先扣钱还是先发货？

分布式事务的经典场景：创建订单 -> 扣库存 -> 扣余额。要是用强一致性，就得用2PC（两阶段提交），但性能差、复杂度高。现实往往采用最终一致性：通过消息队列保证各个操作要么都成功，要么都回滚。

在FastAPI里，我常用fastapi-consumer或直接用Celery处理。核心思想：本地事务 + 消息表 + 重试。

# 伪代码：创建订单时，先写数据库，再发消息
async def create_order(order_data):async with db.transaction():  # 假设支持事务# 1. 插入订单表await db.orders.insert(order_data)# 2. 插入本地消息表await db.messages.insert({"topic": "order_created", "data": order_data, "status": "pending"})# 3. 发送到消息队列（如果失败，有定时任务扫描消息表重发）await send_to_kafka("order_created", order_data)

消费者处理扣库存时，必须保证幂等（用消息ID去重）。万一扣库存失败，可以记录失败，然后人工介入或自动回滚订单（发送补偿消息）。

🎯 这里有个小技巧：用消息表做“发件箱模式”，避免分布式事务，还能保证消息不丢。