使用JWT的REST API并发请求可行性探讨（Python多线程场景）

你的判断没错——如果每个线程各自去获取token，确实会因为新token生成后旧token立即失效，导致先拿到token的线程后续请求失败。但完全可以通过共享token+锁控制的方式实现并发，核心思路是让所有线程复用同一个有效token，只在必要时统一刷新。

具体实现要点：

• 全局共享token与过期时间：维护一个全局变量存储当前有效的JWT，同时记录token的过期时间（比如获取token时计算当前时间+20分钟，甚至可以提前1分钟作为刷新阈值，避免刚好过期时请求失败）。
• 加锁控制刷新逻辑：使用线程锁（比如Python的threading.Lock()）确保同一时间只有一个线程去执行token刷新操作，避免多个线程同时发起认证请求导致重复生成token、失效旧token的问题。
• 请求前校验token状态：每个线程发起业务请求前，先检查当前token是否已过期/即将过期。如果需要刷新，先获取锁，再次确认token状态（防止等待锁的过程中已经被其他线程刷新），然后执行认证请求获取新token，更新全局变量和过期时间后释放锁。

Python多线程示例代码 #

import threading
import time
import requests

# 全局变量：存储当前token和过期时间
current_token = None
token_expire_time = 0
# 线程锁，控制token刷新操作
token_lock = threading.Lock()

# 认证获取token的函数
def get_new_token(username, api_key):
    auth_url = "https://your-auth-endpoint.com/token"
    headers = {"username": username, "api_key": api_key}
    response = requests.post(auth_url, headers=headers)
    response.raise_for_status()
    data = response.json()
    # 返回token和有效期（假设接口返回expires_in为秒数，这里是20分钟=1200秒）
    return data["jwt_token"], time.time() + data["expires_in"]

# 获取可用token的函数（带锁控制）
def get_valid_token(username, api_key):
    global current_token, token_expire_time
    # 先检查是否需要刷新token（提前60秒刷新，避免刚好过期）
    if current_token is None or time.time() > token_expire_time - 60:
        with token_lock:
            # 再次检查，防止等待锁的过程中已经被其他线程刷新了
            if current_token is None or time.time() > token_expire_time - 60:
                current_token, token_expire_time = get_new_token(username, api_key)
    return current_token

# 业务请求函数，每个线程执行这个
def make_api_request(username, api_key, api_url):
    token = get_valid_token(username, api_key)
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get(api_url, headers=headers)
    # 处理响应...
    print(f"请求状态码：{response.status_code}")

# 测试并发
if __name__ == "__main__":
    username = "your_username"
    api_key = "your_api_key"
    api_url = "https://your-api-endpoint.com/data"
    
    # 创建10个线程并发请求
    threads = []
    for _ in range(10):
        t = threading.Thread(target=make_api_request, args=(username, api_key, api_url))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()

关键说明 #

• 锁的作用是确保同一时间只有一个线程去刷新token，避免多个线程同时发起认证请求，导致生成多个token、旧token被频繁失效的问题。
• 两次检查token状态的逻辑（锁外一次，锁内一次）是为了避免“多个线程同时发现token过期，等待锁后重复刷新”的情况——第一个线程刷新完token后，后续线程拿到锁时会发现token已经有效，就不会再重复请求认证接口。

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