Redis缓存与分布式锁在FastAPI高并发场景下的实战优化

1. 为什么需要Redis缓存与分布式锁？

在FastAPI这类现代Web框架中，高并发场景下的性能瓶颈往往出现在数据库IO和资源竞争两个维度。最近处理的一个电商秒杀项目，QPS峰值达到12,000时，MySQL的CPU直接飙到100%，响应时间从50ms恶化到3秒以上。这时候引入Redis就像给系统装上涡轮增压——将商品详情这类热数据缓存后，同样的压力下数据库负载下降了76%。

但缓存不是银弹。当多个请求同时触发缓存重建时，可能引发经典的"缓存击穿"问题。上周我们监控到某个热门商品页在缓存失效瞬间，数据库突然收到238个相同查询。这就是为什么需要分布式锁——它像十字路口的红绿灯，确保在高并发流量下，关键操作依然保持有序。

2. Redis缓存策略深度优化

2.1 多级缓存架构设计

我们采用的缓存策略像俄罗斯套娃：

python复制async def get_product(product_id: str):
    # 第一层：本地内存缓存
    if data := local_cache.get(product_id):
        return data
        
    # 第二层：Redis缓存
    if data := await redis.get(f"product:{product_id}"):
        local_cache.set(product_id, data, 60)  # 本地缓存60秒
        return data
        
    # 第三层：数据库查询
    data = await db.query("SELECT * FROM products WHERE id=?", product_id)
    await redis.setex(f"product:{product_id}", 3600, data)  # Redis缓存1小时
    return data

这种设计使得99%的请求在前两层就被拦截，实测将平均响应时间从87ms降到了23ms。

2.2 缓存失效策略的陷阱

常见的"先删缓存再更新DB"策略在并发场景下可能导致脏数据。我们采用"延迟双删"策略：

python复制async def update_product(product_id: str, new_data: dict):
    await redis.delete(f"product:{product_id}")  # 第一次删除
    await db.update("products", product_id, new_data)
    await asyncio.sleep(1)  # 等待主从同步
    await redis.delete(f"product:{product_id}")  # 第二次删除

这个1秒的延迟是基于我们数据库主从同步的平均耗时计算得出的。

3. 分布式锁的工业级实现

3.1 基于Redlock的锁服务

直接使用SETNX存在锁过期但业务未执行完的风险。我们的实现包含这些关键点：

python复制class DistributedLock:
    def __init__(self, redis, lock_name, timeout=10):
        self.redis = redis
        self.lock_name = lock_name
        self.timeout = timeout
        self.identifier = str(uuid.uuid4())

    async def acquire(self):
        end = time.time() + 5  # 5秒获取超时
        while time.time() < end:
            if await self.redis.set(
                self.lock_name,
                self.identifier,
                ex=self.timeout,
                nx=True
            ):
                return True
            await asyncio.sleep(0.1)
        return False

    async def release(self):
        script = """
        if redis.call("get",KEYS[1]) == ARGV[1] then
            return redis.call("del",KEYS[1])
        else
            return 0
        end
        """
        await self.redis.eval(script, 1, self.lock_name, self.identifier)

这个实现包含三个关键特性：

1. 唯一标识符防止误删其他请求的锁
2. Lua脚本保证原子性检查
3. 自动过期机制避免死锁

3.2 锁续约机制

对于长时间任务，我们增加了看门狗线程：

python复制async def watchdog(lock):
    while lock.held:
        await lock.redis.expire(lock.lock_name, lock.timeout)
        await asyncio.sleep(lock.timeout / 3)

这个续约频率（超时时间的1/3）是在网络延迟和Redis负载间找到的平衡点。

4. 实战中的性能调优

4.1 连接池配置玄机

Redis连接池的配置直接影响并发能力：

python复制import aioredis

redis = await aioredis.create_redis_pool(
    "redis://localhost",
    minsize=5,      # 保持的最小连接数
    maxsize=100,    # 最大连接数
    timeout=3,      # 获取连接超时
    encoding="utf-8"
)

我们通过压力测试发现，当maxsize小于QPS的1/10时，连接等待会成为瓶颈。例如QPS 5000时，连接池至少需要500的大小。

4.2 Pipeline批量操作

在处理排行榜更新时，使用pipeline将20次操作压缩为1次网络往返：

python复制async def update_rankings():
    pipe = redis.pipeline()
    for user_id, score in rankings:
        pipe.zadd("leaderboard", {user_id: score})
    await pipe.execute()

实测这使排行榜更新的吞吐量提升了15倍。

5. 血泪教训：我们踩过的坑

5.1 缓存雪崩事件

某次大促前，我们给1000个商品设置了相同的缓存过期时间。结果在凌晨2点缓存集体失效，数据库瞬间被打垮。现在我们的缓存TTL会加上随机抖动：

python复制ttl = 3600 + random.randint(-300, 300)  # 1小时±5分钟

5.2 锁竞争引发的活锁

当多个服务竞争同一个锁时，可能出现所有请求同时超时又同时重试的死循环。我们通过指数退避算法解决：

python复制base_delay = 0.1
max_delay = 5
while not acquired:
    acquired = await lock.acquire()
    if not acquired:
        delay = min(base_delay * (2 ** retries), max_delay)
        await asyncio.sleep(delay)
        retries += 1

5.3 Redis内存溢出

当缓存大量大对象时，我们遇到过OOM问题。现在会严格监控内存使用，并对大对象进行分块存储：

python复制CHUNK_SIZE = 512 * 1024  # 512KB

async def cache_large_data(key, data):
    chunks = [data[i:i+CHUNK_SIZE] for i in range(0, len(data), CHUNK_SIZE)]
    pipe = redis.pipeline()
    for i, chunk in enumerate(chunks):
        pipe.set(f"{key}:{i}", chunk)
    await pipe.execute()

6. 监控指标体系建设

完善的监控能提前发现问题。我们收集这些关键指标：

通过Grafana配置的看板可以直观看到缓存系统的健康状态。当缓存命中率低于阈值时，会触发自动扩容机制。