FastAPI中如何避免并发请求重复刷新Bearer Token？

场景与示例代码 #

我的FastAPI接口需要调用受保护的外部API，请求必须携带Bearer Token。应用逻辑会先检查当前Token是否过期，若过期则获取新Token。示例代码如下：

from fastapi import FastAPI

app = FastAPI()

class Token:
    def expired(self) -> bool:
        ...

    def __str__(self) -> str:
        ...

def get_token() -> Token:
    ...

async def issue_new_token() -> Token:
    ...

async def touch_external_api(headers: dict):
    ...

@app.post('/process/{task_id}')
async def handler(task_id):
    token = get_token()
    if token.expired():
        token = await issue_new_token()
    await touch_external_api(headers={'Authorization': f'Bearer {token}'})

核心问题 #

当多个并发请求触发该接口，且token.expired()返回True时，如何确保只有第一个请求调用issue_new_token()，其余请求等待新Token生成完成后直接复用？或者说怎么保证新Token只被获取一次？另外，用后台任务预先更新Token的方案是否可行？

我的尝试（未测试） #

我考虑用asyncio的锁原语实现，但感觉不够优雅，代码如下：

import asyncio

token_lock = asyncio.Lock()

async def issue_new_token() -> Token:
    ...

async def issue_token() -> Token:
    if not token_lock.locked():
        await token_lock.acquire()
        token = await issue_new_token()
        token_lock.release()
    else:
        await token_lock.acquire()  # wait until coroutine which issues new token releases
        token_lock.release()
        token = get_token()
    return token

@app.post('/process/{task_id}')
async def handler(task_id):
    token = get_token()
    if token.expired():
        token = await issue_token()
    await touch_external_api(headers={'Authorization': f'Bearer {token}'})

可行方案一：优化asyncio锁实现（解决并发刷新问题） #

你的思路方向正确，但原代码存在两处关键问题：一是token_lock.locked需改为方法调用token_lock.locked()；二是缺少二次检查——在获取锁后必须再次判断Token是否过期，避免等待锁的过程中已有其他请求完成刷新。

优化后的代码更简洁且健壮：

import asyncio

token_lock = asyncio.Lock()

async def issue_new_token() -> Token:
    # 实现调用外部接口获取新Token的逻辑，并更新全局Token存储
    new_token = ...
    save_token(new_token)  # 假设此函数将新Token存入全局存储，供get_token()读取
    return new_token

async def get_or_refresh_token() -> Token:
    current_token = get_token()
    if not current_token.expired():
        return current_token
    
    async with token_lock:
        # 二次检查：获取锁后再次确认Token状态
        current_token = get_token()
        if current_token.expired():
            current_token = await issue_new_token()
        return current_token

@app.post('/process/{task_id}')
async def handler(task_id):
    token = await get_or_refresh_token()
    await touch_external_api(headers={'Authorization': f'Bearer {token}'})

这里用async with替代手动管理锁的生命周期，能确保锁在异常场景下也会被释放。二次检查是核心逻辑：当多个请求同时检测到Token过期并等待锁时，第一个拿到锁的请求会执行刷新，后续请求获取锁后会发现Token已有效，直接返回即可，彻底避免重复调用issue_new_token()。

可行方案二：后台任务预更新Token（从根源避免并发场景） #

后台预更新方案完全可行，适合Token有效期较长的场景。思路是启动一个定时后台任务，定期检查Token剩余有效期，当Token即将过期（比如剩余10%有效期）时提前刷新，确保用户请求到来时Token始终有效。

实现示例：

from fastapi import FastAPI
import asyncio
from contextlib import asynccontextmanager

app = FastAPI()
token_refresh_task: asyncio.Task | None = None

async def token_refresh_background_task():
    while True:
        current_token = get_token()
        # 假设Token总有效期3600秒，剩余不足360秒时触发刷新
        if current_token.expires_in() < 360:
            await issue_new_token()
        # 每隔60秒检查一次
        await asyncio.sleep(60)

@asynccontextmanager
async def lifespan(app: FastAPI):
    global token_refresh_task
    # 启动后台刷新任务
    token_refresh_task = asyncio.create_task(token_refresh_background_task())
    yield
    # 应用关闭时取消任务
    token_refresh_task.cancel()
    try:
        await token_refresh_task
    except asyncio.CancelledError:
        pass

app.router.lifespan_context = lifespan

# Token类、get_token、issue_new_token等其余代码保持不变

@app.post('/process/{task_id}')
async def handler(task_id):
    token = get_token()
    # 保留兜底检查，防止后台任务异常导致Token过期
    if token.expired():
        token = await issue_new_token()
    await touch_external_api(headers={'Authorization': f'Bearer {token}'})

该方案的优势是从根源上消除并发刷新的可能，但需注意：

• 检查间隔和提前刷新阈值需根据Token实际有效期调整
• 若为分布式部署（多实例），每个实例会独立运行后台任务，可能导致重复刷新。这种情况需结合分布式锁（如Redis锁）保证只有一个实例执行刷新操作。

内容的提问来源于stack exchange，提问作者hardhypochondria