Python中处理JWT并发请求时Token覆盖问题有什么好的解决方法?

如:在一个页面同时异步请求两个接口,这时 accesstoken 过期了,然后第一个请求去请求 refreshtoken ,后端覆盖了 accesstoken,第二个请求就出现 token 无效,要求重新登录。。。。


Python中处理JWT并发请求时Token覆盖问题有什么好的解决方法?
18 回复

当后端返回 accesstoken 过期时,中断所有请求,调用一个不能并发执行的方法刷新令牌,令牌刷新完后弄个回调函数重新去请求接口


这个问题在微服务架构里挺常见的。核心是多个并发请求共享同一个JWT token对象时,一个请求刷新token后会影响其他正在进行的请求。

我通常用请求级别的token隔离来解决。下面是个完整的示例:

import threading
from functools import wraps
import requests
from datetime import datetime, timedelta

class TokenManager:
    def __init__(self):
        self._tokens = threading.local()  # 线程局部存储
        self._refresh_lock = threading.Lock()
    
    def get_token(self):
        """获取当前请求的token"""
        if not hasattr(self._tokens, 'current_token'):
            self._refresh_token()
        return self._tokens.current_token
    
    def _refresh_token(self):
        """刷新token(带锁防止并发刷新)"""
        with self._refresh_lock:
            # 模拟获取新token的逻辑
            new_token = f"token_{datetime.now().timestamp()}"
            self._tokens.current_token = new_token
            return new_token

# 使用装饰器确保每个请求有独立的token上下文
def token_context(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 每个请求线程初始化自己的token存储
        if not hasattr(token_manager._tokens, 'current_token'):
            token_manager._tokens.current_token = None
        return func(*args, **kwargs)
    return wrapper

token_manager = TokenManager()

@token_context
def api_request(url):
    """模拟API请求"""
    token = token_manager.get_token()
    # 使用token发起请求...
    print(f"Request to {url} with token: {token}")
    return {"status": "success"}

# 模拟并发请求
import concurrent.futures

def make_concurrent_requests():
    urls = ["/api/user", "/api/order", "/api/product"]
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        futures = [executor.submit(api_request, url) for url in urls]
        results = [future.result() for future in futures]
    
    return results

if __name__ == "__main__":
    results = make_concurrent_requests()

关键点:

  1. threading.local()为每个线程创建独立的token存储
  2. 加锁控制token刷新,避免重复刷新
  3. 装饰器确保每个请求初始化自己的上下文

这样每个请求线程操作的都是自己的token副本,不会互相干扰。如果用了异步框架(比如FastAPI),可以把threading.local()换成contextvars

简单说就是线程隔离加锁刷新。

查了一下资料,方案: 给 accesstoken 一个过期时间,然后每次请求验证 是否过期,当要过期时(离过期还有十分钟之内的)就去请求新的 token,然后后端 把旧的 token 缓存着,例如缓存十分钟,十分钟之内,使用旧的 token 依然可以请求。

可以参考 jwt-auth 的做法, 可以配置一个并发失效时间, 比如 2s,
意思是即使这个 token 过期了, 那么这个 token 也能在 2s 内使用.

如果是 oauth2,一般框架会在返回令牌的时候给个过期时间,可以根据这个做提前处理

话说回来,刷新令牌多刷几个又能怎么样……为什么要对访问令牌做废弃检查

因为存在数据库了 每次验证 都进行对比了 多刷的令牌 导致后面旧的令牌无法听过验证。。。。

我去看看 我用的 flask-jwt-extended 看看有没有这个配置。。。

请求 token 的操作由统一的 proxy 来负责,如果已经有请求发出去,则不再重复发送。

jwt 还要验证失效,那还不如随机 token。

如果没有这个配置, 可以自己设置, 也比较简单逻辑.
建立一个中间件. 然后解析出 token 的有效信息. 得到 exp. 根据当前时间和 exp 对比, 如果过期在 2s 之内, 都允许通过.

JWT 是不储存过期时间在服务端的,过期时间储存在 token 里面,不能修改,所以要提前实战只能设置 black list


我的意思是多发几个令牌对于你们业务有没有什么影响
jwt 的令牌应该是签出去就有效,如果有废弃的需求,应该从令牌容纳的数据去设计,而不是废弃令牌本身

宽限时间了解一下

jwt 的意义就是不存后端也不直接废弃。如果你们存后端还要主动废弃,也不知道是想利用 jwt 的哪种特性。

他们以前就这样用的 差不多 就是单一登录了。。。。。。

了解了一下 非常好,但是我用哪个插件似乎没有这个功能 要手动实现 。。。

找到配置了 。。。

回到顶部