如何阻止刷新认证令牌的后端方法因多API请求重复执行

这个问题的核心是并发请求导致的令牌更新顺序混乱——先发起的请求因为响应慢，反而覆盖了后发起请求返回的有效令牌。我们可以通过互斥锁+双重检查的方式来解决，既保证同一时间只有一个刷新请求在执行，又避免不必要的重复刷新。

方案一：同步场景下的线程锁（适用于requests同步调用） #

首先在你的类中初始化一个线程锁，用来控制刷新操作的串行执行：

import threading
import time

class YourAuthClass:
    def __init__(self):
        # 初始化互斥锁
        self.refresh_lock = threading.Lock()
        # 记录最后一次刷新时间，避免频繁重复刷新
        self.last_refresh_timestamp = 0
        # 设置冷却时间（比如60秒内不重复刷新，根据你的令牌过期时间调整）
        self.refresh_cooldown = 60
        # 保存当前有效的令牌
        self.current_refresh_token = None
        self.current_auth_token = None

    def is_token_valid(self):
        # 这里实现你的令牌有效性判断逻辑（比如检查是否过期）
        # 示例：假设当前令牌存在且未过期
        return self.current_auth_token is not None

    def get_current_refresh_token(self):
        # 返回当前保存的最新刷新令牌
        return self.current_refresh_token

    def get_refresh_token(self):
        current_time = time.time()

        # 第一步：先检查是否在冷却期内且令牌有效，直接返回现有令牌
        if current_time - self.last_refresh_timestamp < self.refresh_cooldown and self.is_token_valid():
            return self.get_current_refresh_token()

        # 第二步：尝试获取锁，控制刷新操作串行执行
        # 先尝试非阻塞获取锁，如果失败说明已有请求在刷新
        if not self.refresh_lock.acquire(blocking=False):
            # 这里可以选择等待锁释放后再处理（保证拿到最新令牌）
            with self.refresh_lock:
                # 双重检查：等待锁期间可能已经完成了刷新
                if current_time - self.last_refresh_timestamp < self.refresh_cooldown and self.is_token_valid():
                    return self.get_current_refresh_token()
                # 执行刷新逻辑
                body = "token=123LastRefreshToken"
                response = requests.post(url, headers, data=body)
                self.update_auth(response)
                new_token = self.save_new_refesh_token(response)
                self.last_refresh_timestamp = current_time
                return new_token
        else:
            # 成功获取锁，执行刷新逻辑
            try:
                body = "token=123LastRefreshToken"
                response = requests.post(url, headers, data=body)
                self.update_auth(response)
                new_token = self.save_new_refesh_token(response)
                self.last_refresh_timestamp = current_time
                return new_token
            finally:
                # 无论请求成功失败，都要释放锁
                self.refresh_lock.release()

    def update_auth(self, response):
        # 你的现有逻辑：更新认证信息
        pass

    def save_new_refesh_token(self, response):
        # 你的现有逻辑：保存新的刷新令牌，并返回
        new_token = response.json().get("refresh_token")
        self.current_refresh_token = new_token
        self.current_auth_token = response.json().get("auth_token")
        return new_token

关键逻辑说明： #

• 互斥锁：threading.Lock确保同一时间只有一个线程执行刷新请求，彻底避免并发刷新导致的令牌覆盖。
• 双重检查（DCL）：在获取锁前后都检查令牌有效性，避免等待锁期间已经完成刷新，减少不必要的请求。
• 冷却时间：前端每秒多次调用，很多时候不需要刷新，通过冷却时间过滤重复请求，提升后端性能。

方案二：异步场景下的异步锁（适用于FastAPI/aiohttp等异步框架） #

如果你的后端是异步架构，需要使用asyncio.Lock来替代线程锁，代码结构类似，但要适配异步语法：

import asyncio
import time
import aiohttp

class YourAsyncAuthClass:
    def __init__(self):
        self.refresh_lock = asyncio.Lock()
        self.last_refresh_timestamp = 0
        self.refresh_cooldown = 60
        self.current_refresh_token = None
        self.current_auth_token = None

    def is_token_valid(self):
        return self.current_auth_token is not None

    def get_current_refresh_token(self):
        return self.current_refresh_token

    async def get_refresh_token(self):
        current_time = time.time()

        if current_time - self.last_refresh_timestamp < self.refresh_cooldown and self.is_token_valid():
            return self.get_current_refresh_token()

        # 异步锁的上下文管理
        async with self.refresh_lock:
            # 双重检查
            if current_time - self.last_refresh_timestamp < self.refresh_cooldown and self.is_token_valid():
                return self.get_current_refresh_token()
            
            # 用aiohttp发起异步请求
            async with aiohttp.ClientSession() as session:
                body = "token=123LastRefreshToken"
                async with session.post(url, headers=headers, data=body) as response:
                    response_data = await response.json()
                    self.update_auth(response_data)
                    new_token = self.save_new_refesh_token(response_data)
                    self.last_refresh_timestamp = current_time
                    return new_token

    def update_auth(self, response_data):
        # 更新认证信息的异步逻辑（如果需要的话）
        pass

    def save_new_refesh_token(self, response_data):
        new_token = response_data.get("refresh_token")
        self.current_refresh_token = new_token
        self.current_auth_token = response_data.get("auth_token")
        return new_token

额外注意事项 #

• 请求失败处理：如果刷新请求失败（比如网络错误、服务端返回错误），要确保锁能正常释放，避免死锁。上面的代码用try-finally（同步）和async with（异步）已经处理了这一点。
• 令牌有效性判断：一定要实现可靠的is_token_valid方法，比如检查令牌的过期时间，而不是只依赖冷却时间，这样能更准确地控制刷新时机。

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