多Worker多线程异步FastAPI应用中Redis缓存Token的并发刷新锁实现咨询

嗨，我完全理解你现在的困扰——在多Worker、多线程的异步FastAPI环境里，Token缓存的并发刷新确实很容易踩坑，之前用MySQL存Token加asyncio.Lock的方案，确实没法覆盖跨Worker的场景，而且长期占着数据库连接也不符合最佳实践。换成Redis是非常靠谱的选择，我来给你详细讲讲怎么实现正确的分布式锁，解决并发刷新的问题：

核心思路：用Redis的原子性SET命令实现分布式锁 #

Redis是单线程执行命令的，所以我们可以利用SET命令的NX（只有当key不存在时才设置）参数，配合过期时间，实现一个跨Worker、跨线程的分布式锁。这个操作是原子性的，能保证同一时间只有一个请求能拿到锁，去执行Token刷新操作。

异步场景下的具体实现（用aioredis） #

因为你的FastAPI是异步的，所以推荐用aioredis这个异步Redis客户端，下面是完整的示例代码：

import asyncio
import aioredis
from fastapi import FastAPI

# 初始化FastAPI应用和Redis连接
app = FastAPI()
redis = aioredis.from_url("redis://localhost", decode_responses=True)

async def get_external_api_token():
    # 第一步：先尝试从Redis获取已缓存的有效Token
    cached_token = await redis.get("external_api_token")
    if cached_token:
        return cached_token

    # 第二步：尝试获取刷新锁，锁的过期时间设为比Token有效期短（比如5分钟），防止锁永久占用
    lock_acquired = await redis.set(
        "token_refresh_lock", 
        "locked", 
        nx=True,  # 只有key不存在时才成功
        ex=300    # 锁5分钟后自动过期
    )
    
    if not lock_acquired:
        # 没拿到锁，说明有其他请求正在刷新Token，稍等片刻后重试
        await asyncio.sleep(0.1)
        return await get_external_api_token()

    try:
        # 再次检查Token，避免等待锁的过程中已经被其他请求刷新完成
        cached_token = await redis.get("external_api_token")
        if cached_token:
            return cached_token

        # 第三步：调用外部API获取新Token
        new_token = await fetch_new_token_from_api()
        
        # 第四步：将新Token存入Redis，设置和Token有效期一致的过期时间
        await redis.set("external_api_token", new_token, ex=3600)  # 假设Token有效期1小时
        return new_token
    finally:
        # 无论刷新成功或失败，都释放锁（如果进程崩溃，锁的过期时间会兜底）
        await redis.delete("token_refresh_lock")

async def fetch_new_token_from_api():
    # 这里替换成你调用外部API获取Token的实际逻辑
    await asyncio.sleep(1)  # 模拟API请求耗时
    return "new_valid_token_12345"

# 示例接口，演示如何使用Token获取函数
@app.get("/use-token")
async def use_token():
    token = await get_external_api_token()
    return {"token": token, "message": "Token fetched successfully"}

关键细节说明 #

• 双重检查Token：在拿到锁之后再次检查Redis中的Token，是为了避免在等待锁的短暂时间里，已经有其他请求完成了刷新，这样可以避免重复刷新。
• 锁的过期时间兜底：设置锁的过期时间，是防止拿到锁的请求因为意外（比如进程崩溃）无法释放锁，导致后续所有请求都无法刷新Token。
• 异步重试逻辑：没拿到锁的请求通过短暂睡眠后重试，既不会给Redis造成太大压力，也能保证最终拿到最新的Token。

进阶优化：处理长耗时的Token刷新 #

如果你的外部API响应很慢，刷新Token的时间可能超过锁的过期时间，这时候可以加一个“锁看门狗”机制，在刷新过程中定期延长锁的过期时间：

async def refresh_lock_watchdog(lock_key: str, expire_seconds: int):
    while True:
        await asyncio.sleep(10)  # 每10秒续一次锁
        # 只有锁还存在时才延长过期时间
        await redis.expire(lock_key, expire_seconds)

# 在get_external_api_token的try块中添加：
try:
    # 启动看门狗任务，定期续锁
    watchdog_task = asyncio.create_task(refresh_lock_watchdog("token_refresh_lock", 300))
    # ... 原有刷新Token的逻辑 ...
finally:
    # 取消看门狗任务并释放锁
    watchdog_task.cancel()
    await redis.delete("token_refresh_lock")

为什么这个方案适合你的场景 #

• 跨Worker/线程生效：Redis的分布式锁不受进程和线程限制，不管你有多少Worker或线程，都能保证只有一个请求执行Token刷新。
• 符合异步最佳实践：所有Redis操作都是异步的，不会阻塞事件循环，也不需要长期占用数据库连接。
• 原子性保障：Redis的SET nx ex命令是原子执行的，不会出现多个请求同时拿到锁的情况。

备注：内容来源于stack exchange，提问作者SteveGu