如何在Locust中实现仅执行一次Auth Token请求并复用？

你遇到的问题是：每个虚拟用户实例启动时都会执行on_start方法，导致并发多少用户就会发起多少次Token请求。要实现全局仅获取一次Token并复用，需要将Token的获取逻辑改为全局初始化，而非每个用户单独执行。

解决方案1：全局缓存Token+线程锁（推荐） #

通过全局变量存储Token，配合线程锁确保仅第一次请求获取Token，后续所有用户复用，同时自动处理Token过期刷新：

from locust import HttpUser, constant, task
import urllib3
import time
import requests
from threading import Lock

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# 全局存储Token、过期时间，加锁避免并发重复请求
_global_token = None
_token_expiry = 0
_token_lock = Lock()

def fetch_token():
    global _global_token, _token_expiry
    current_time = time.time()
    
    # 检查Token是否有效（提前60秒刷新，避免过期）
    if _global_token and current_time < _token_expiry - 60:
        return _global_token
    
    # 加锁确保同一时间只有一个请求获取Token
    with _token_lock:
        if _global_token and current_time < _token_expiry - 60:
            return _global_token
        
        # 构造Token请求参数
        tenant_id = "tenant123"
        client_id = "client123"
        secret = "secret123"
        scope = "api://123/.default"
        body = f"grant_type=client_credentials&client_id={client_id}&client_secret={secret}&scope={scope}"
        
        # 用requests直接发起请求（避免关联到Locust用户实例的请求统计）
        response = requests.post(
            f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
            data=body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            verify=False
        )
        response.raise_for_status()
        token_data = response.json()
        
        _global_token = token_data['access_token']
        _token_expiry = current_time + token_data['expires_in']
        return _global_token

class ProcessRequests(HttpUser):
    host = 'https://hostURL'
    wait_time = constant(1)

    def on_start(self):
        # 从全局缓存获取Token，设置到当前用户的请求头
        token = fetch_token()
        self.headers = {'Authorization': f'Bearer {token}'}

    @task
    def get_labware(self):
        self.client.get("/123", name="Labware", headers=self.headers)

    @task
    def get_instruments(self):
        self.client.get("/456", name="Instruments", headers=self.headers)

解决方案2：利用Locust测试启动事件初始化Token #

通过events.test_start事件，在测试开始时仅执行一次Token获取，所有用户直接复用：

from locust import HttpUser, constant, task, events
import urllib3
import time
import requests

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

_global_token = None
_token_expiry = 0

# 测试启动时执行一次Token获取
@events.test_start.add_listener
def init_token(environment, **kwargs):
    global _global_token, _token_expiry
    tenant_id = "tenant123"
    client_id = "client123"
    secret = "secret123"
    scope = "api://123/.default"
    body = f"grant_type=client_credentials&client_id={client_id}&client_secret={secret}&scope={scope}"
    
    response = requests.post(
        f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
        data=body,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        verify=False
    )
    response.raise_for_status()
    token_data = response.json()
    
    _global_token = token_data['access_token']
    _token_expiry = time.time() + token_data['expires_in']

class ProcessRequests(HttpUser):
    host = 'https://hostURL'
    wait_time = constant(1)

    def on_start(self):
        # 若测试时长超过Token有效期，可在此处添加过期刷新逻辑
        current_time = time.time()
        if current_time >= _token_expiry - 60:
            # 重复上述Token请求逻辑，或调用封装好的fetch_token函数
            pass
            
        self.headers = {'Authorization': f'Bearer {_global_token}'}

    @task
    def get_labware(self):
        self.client.get("/123", name="Labware", headers=self.headers)

    @task
    def get_instruments(self):
        self.client.get("/456", name="Instruments", headers=self.headers)

关键说明 #

• 原代码的on_start是每个虚拟用户实例的初始化方法，并发用户数越多，Token请求次数越多；
• 两种方案均通过全局变量存储Token，确保仅执行一次获取请求；
• 加入了Token过期检查逻辑，避免测试过程中Token失效导致请求失败。

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