1. 项目概述:为什么大模型API并发测试是门“必修课”
最近在折腾几个AI应用项目,从智能客服到内容生成,核心都绕不开调用大模型API。项目上线前,团队里总会有人问:“咱们这系统,到底能扛住多少用户同时提问?” 这个问题看似简单,背后却是一连串的技术焦虑:API调用会不会突然变慢?服务商有没有限流?我们的钱包能不能撑住突发的流量洪峰?更重要的是,用户体验会不会在关键时刻“掉链子”?这些问题,靠猜是没用的,必须靠数据说话。这就是为什么我们需要对大模型API进行并发压力测试,而Locust正是我用来解决这个问题的“瑞士军刀”。
你可能听说过JMeter、wrk这些压测工具,为什么我独爱Locust?原因很简单:Python原生。大模型API的调用逻辑,无论是使用OpenAI格式、还是国内智谱、DeepSeek的SDK,本质上都是一段Python代码。用Locust,我可以用写业务逻辑的思维来设计压测场景,模拟真实的用户思考、等待、连续对话等复杂行为,而不是在图形界面里拖拽一个个难以维护的元件。它能精准地告诉我,在每秒50个、100个甚至500个并发请求下,我的API响应时间(P99延迟)是多少,失败率有多高,从而清晰地找到系统的性能瓶颈和成本临界点。
这篇文章,我就以一个真实的“智能问答系统”接入某大模型API的场景为例,带你从零开始,用Locust完成一次完整的、可复现的并发极限评估实战。你会学到如何设计贴近真实的用户行为脚本、如何解读关键的压测指标、以及如何避开那些我踩过的“坑”。无论你是后端开发、算法工程师还是项目负责人,这套方法都能帮你把“系统能扛多少并发”从一个模糊的疑问,变成一个精确的、有数据支撑的答案。
2. 核心思路与工具选型:为什么是Locust+异步?
在规划这次压测时,我首先明确了几个核心目标:第一,测试场景必须高度模拟真实用户,不能是简单的单次请求循环;第二,工具要能清晰展示性能瓶颈,是网络、API限流还是我自身代码的问题;第三,整个流程要易于集成和自动化,方便在每次API更新或业务逻辑调整后快速回归测试。基于这三点,Locust几乎是唯一的选择。
2.1 Locust的核心优势解析
Locust是一个用Python编写的开源负载测试工具。它的核心哲学是“用代码定义用户行为”,这带来了几个无可替代的优势:
- 测试即代码:你的压测脚本(locustfile.py)就是一个标准的Python模块。你可以使用requests、aiohttp、openai等任何你熟悉的库来发起请求,也可以方便地引入业务逻辑,比如先登录获取token,再发起对话。这使得测试脚本的维护和版本控制变得极其简单。
- 分布式与可扩展性:单机跑不动了?Locust原生支持分布式运行。一台机器作为主节点(master),多台机器作为从节点(worker),可以轻松模拟数万甚至数十万的并发用户。这对于真正想探知大模型API全局限流阈值的情况至关重要。
- 实时Web UI与详实数据:启动Locust后,可以通过浏览器访问一个本地Web界面,实时查看当前RPS(每秒请求数)、响应时间、失败率等关键指标。测试结束后,还能生成详细的HTML报告,方便团队分享和归档。
- 资源消耗相对较低:相比于一些基于Java的GUI工具,Locust(特别是使用异步客户端时)在资源利用上更为高效,单机也能模拟较高的并发。
2.2 同步与异步客户端的关键抉择
这是使用Locust时第一个重要的技术决策点。Locust支持两种HTTP客户端:默认的同步HttpUser和基于gevent的FastHttpUser,以及需要手动实现的异步客户端(如aiohttp)。
- 同步客户端(HttpUser):使用简单,适合请求间隔较长或逻辑简单的场景。但在模拟高并发时,每个虚拟用户(User)都是一个独立的greenlet(微线程),虽然比操作系统线程轻量,但在发起网络IO等待时,仍然会进行上下文切换。当并发数极高时(例如数千),其调度开销会变得明显。
- 异步客户端(aiohttp + asyncio):这是我强烈推荐用于大模型API压测的方案。大模型API的响应时间通常在1-10秒甚至更长,这是一个典型的高延迟、低频率的IO密集型场景。使用异步,可以用极少的操作系统线程(甚至一个)管理成千上万个并发连接。每个虚拟用户在一个事件循环中发起请求后便挂起,让出控制权给其他用户,直到收到响应再恢复。这能极大地提升单机模拟高并发的上限,并且更准确地反映真实世界中大量用户同时等待长响应的场景。
简单来说,如果你想在个人电脑上就模拟出上千用户同时与大模型对话的场景,异步方案是必由之路。接下来的实战部分,我将基于异步方案展开。
2.3 大模型API压测的特殊性
与测试一个返回“Hello World”的普通HTTP接口不同,测试大模型API需要特别注意以下几点:
- 令牌(Token)管理:API Key是敏感信息,不能硬编码在脚本中。需要安全地通过环境变量或外部配置文件传入。
- 请求体构造:大模型请求的Prompt、参数(如max_tokens, temperature)会显著影响响应时间和token消耗,从而影响性能和成本。压测时应使用有代表性的、符合业务场景的Prompt。
- 响应解析与断言:除了检查HTTP状态码是否为200,我们还需要验证响应体结构是否正确、是否包含了预期的“choices”或“message”字段,甚至可以对返回的文本内容做简单校验(如是否非空)。
- 尊重服务商限流:盲目地进行极限压测可能导致你的API Key被临时封禁。务必先从低并发开始,阶梯式增加,并密切关注返回的429(Too Many Requests)或其他限流错误。我们的目标是找到稳定运行的“甜蜜点”,而不是暴力攻击。
3. 环境准备与Locustfile核心脚本编写
理论说得再多,不如一行代码。让我们开始动手搭建压测环境。我假设你已经在本地或测试服务器上准备好了Python环境(3.7+)。
3.1 依赖安装与项目结构
首先,创建一个干净的目录,并安装必要的包。我们主要需要locust和异步HTTP客户端aiohttp。
# 创建项目目录 mkdir mlloadtest && cd mlloadtest # 创建虚拟环境(推荐) python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装核心依赖 pip install locust aiohttp # 可选:安装你所用大模型官方的SDK,例如OpenAI、智谱等,方便构造请求。 # pip install openai项目目录结构可以这样规划:
mlloadtest/ ├── locustfile.py # 核心压测脚本 ├── prompts.json # 存储测试用的Prompt池 ├── .env # 存储API Key等环境变量(务必加入.gitignore) └── requirements.txt # 依赖列表3.2 编写异步Locustfile脚本
这是整个压测的核心。我们将创建一个继承自HttpUser但使用aiohttp会话的异步用户类。
# locustfile.py import os import random import asyncio from locust import HttpUser, task, between, events from locust.exception import StopUser import aiohttp import json from dotenv import load_dotenv # 用于加载.env文件 # 加载环境变量 load_dotenv() class AsyncAIOHttpSession: """一个简单的aiohttp会话包装器,用于在Locust中记录请求统计。""" def __init__(self, user): self.user = user self.session = None async def __aenter__(self): self.session = aiohttp.ClientSession() return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self.session.close() async def post(self, url, **kwargs): """发起POST请求,并自动集成到Locust的统计中。""" # 为请求命名,用于Locust报告中的区分 name = kwargs.pop('name', url) start_time = time.time() response = None try: response = await self.session.post(url, **kwargs) # 手动触发Locust的成功/失败事件 total_time = int((time.time() - start_time) * 1000) # 转毫秒 if response.status < 400: events.request.fire( request_type="POST", name=name, response_time=total_time, response_length=len(await response.text()), context=self.user._context(), exception=None, ) else: events.request.fire( request_type="POST", name=name, response_time=total_time, response_length=0, context=self.user._context(), exception=Exception(f"HTTP Error {response.status}"), ) return response except Exception as e: total_time = int((time.time() - start_time) * 1000) events.request.fire( request_type="POST", name=name, response_time=total_time, response_length=0, context=self.user._context(), exception=e, ) raise # 假设我们测试一个兼容OpenAI格式的API端点 API_BASE_URL = os.getenv("API_BASE_URL", "https://api.example.com/v1") API_KEY = os.getenv("API_KEY") MODEL_NAME = os.getenv("MODEL_NAME", "gpt-3.5-turbo") # 加载测试Prompt池 with open('prompts.json', 'r', encoding='utf-8') as f: PROMPT_POOL = json.load(f)['prompts'] class ChatAPIUser(HttpUser): """ 模拟一个与大模型进行对话的用户。 每个用户会随机从Prompt池中选择一个问题进行提问。 """ # 用户任务之间的等待时间范围(秒),模拟用户思考时间。 wait_time = between(2, 5) def on_start(self): """当虚拟用户启动时执行,可用于初始化会话、登录等。""" # 初始化aiohttp会话 self.aiohttp_session = None # 注意:由于Locust的架构,我们需要在任务中异步初始化,这里先占位。 @task def ask_question(self): """ 核心任务:向大模型API发送一个提问请求。 由于Locust的@task默认不支持async,我们通过asyncio.run来运行异步函数。 在高版本Locust或特定模式下,可以考虑使用@task(weight)装饰异步函数。 这里采用一个兼容性较好的模式:在同步任务中运行异步事件循环。 """ # 为每个任务创建一个新的事件循环(在非WebUI模式下可行) loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: loop.run_until_complete(self._async_ask_question()) finally: loop.close() async def _async_ask_question(self): """真正的异步请求逻辑""" if self.aiohttp_session is None: self.aiohttp_session = AsyncAIOHttpSession(self) async with self.aiohttp_session as session: # 1. 随机选择一个Prompt prompt = random.choice(PROMPT_POOL) # 2. 构造请求体 (OpenAI格式示例) payload = { "model": MODEL_NAME, "messages": [{"role": "user", "content": prompt}], "max_tokens": 150, # 控制生成长度,影响响应时间和成本 "temperature": 0.7, } headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" } # 3. 发起请求,并使用‘name’参数标识此请求类型 url = f"{API_BASE_URL}/chat/completions" try: response = await session.post( url, name="/chat/completions", json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=30) # 设置超时 ) # 4. 检查响应 if response.status == 200: data = await response.json() # 可选:验证响应结构 if not data.get("choices"): raise ValueError("Response missing 'choices' field") # 你可以在这里对回复内容做进一步检查,例如长度、关键词等 # print(f"User got reply: {data['choices'][0]['message']['content'][:50]}...") elif response.status == 429: # 遇到限流,可以记录日志,并让这个虚拟用户“休息”一会儿 print(f"Rate limited! Status: {response.status}") await asyncio.sleep(5) # 等待5秒后重试或继续 else: # 其他错误,记录并可能停止该用户(根据测试策略) error_text = await response.text() print(f"Request failed with status {response.status}: {error_text}") except asyncio.TimeoutError: print("Request timed out!") except Exception as e: print(f"An error occurred: {e}")脚本关键点解析与避坑指南:
- 异步会话封装:
AsyncAIOHttpSession类是我们的核心创新点。它包装了aiohttp.ClientSession,并在每个请求前后手动触发了Locust的events.request事件。这是将异步请求集成到Locust统计系统的关键。没有这一步,Locust的Web UI将无法正确显示你的请求数据。 - 任务(@task)装饰器:Locust原生的
@task装饰器不支持async def。我们采用了一个变通方案:在同步的ask_question方法内,使用asyncio.run或loop.run_until_complete来执行真正的异步函数_async_ask_question。在高并发下,为每个任务创建新的事件循环有一定开销,但对于大模型API这种长耗时请求,这个开销可以接受。更优雅的方案是使用locust-plugins等第三方库的异步支持,但为了减少依赖,本例采用基础方法。 - 超时设置:
aiohttp.ClientTimeout(total=30)至关重要。大模型API响应可能很慢,但没有超时设置的请求会一直挂起,耗尽系统资源。设置一个合理的总超时(如30秒),超时后触发asyncio.TimeoutError,Locust会将其记录为失败请求。 - 错误处理与限流:我们特别处理了HTTP 429状态码(限流)。在真实压测中,一旦频繁出现429,意味着已经触达API提供方的当前限制。此时不应继续疯狂重试,而是应该让虚拟用户“睡眠”一下,或者停止增加并发数,这更符合真实场景和“友好测试”的原则。
- Prompt池:将测试用的Prompt放在外部的
prompts.json文件中,使得测试用例可以灵活扩展,更贴近真实用户输入的多样性。文件内容类似:{ "prompts": [ "用简单的语言解释一下什么是机器学习?", "写一首关于春天的五言绝句。", "计算一下15的阶乘是多少?", "总结《三国演义》中赤壁之战的主要经过。", "将‘Hello, world!’翻译成法语和西班牙语。" ] }
4. 压测执行策略与关键参数配置
脚本准备好了,接下来是如何执行它。Locust提供了命令行和Web UI两种交互方式,对于探索性测试和结果展示,Web UI非常直观;对于自动化或CI/CD集成,命令行模式更合适。
4.1 通过Web UI执行与监控
这是最常用的方式,适合交互式地调整负载,观察实时曲线。
# 在项目根目录下执行 locust -f locustfile.py启动后,打开浏览器访问http://localhost:8089,你会看到Locust的Web控制台。
控制台参数配置详解:
- Number of users (peak concurrency):峰值并发用户数。这是Locust中最重要的概念之一。它不等于每秒请求数(RPS)。例如,设置1000个用户,且每个用户平均每5秒执行一个任务(
wait_time = between(2, 5)),那么理论上的RPS约为 1000 users / 5s = 200 RPS。我们的目标是找到系统稳定运行下能支持的“最大峰值并发用户数”。 - Spawn rate (users started/second):孵化率,即每秒启动多少个虚拟用户,直到达到峰值用户数。设置为10,意味着每秒增加10个用户,慢慢给系统加压,而不是瞬间将所有用户砸向系统。这有助于观察系统负载逐渐上升时的表现,是一种更温和、更真实的测试方式。
- Host:这里填写你的大模型API的基础URL,例如
https://api.example.com/v1。注意,我们的脚本里已经定义了API_BASE_URL,如果这里也填写了,脚本中的URL会以此为准。
执行流程建议:
- 第一步:冒烟测试。设置
Number of users = 1,Spawn rate = 1,运行1分钟。确保单个用户能正常请求并得到响应,验证脚本和环境无误。 - 第二步:阶梯加压。这是最关键的步骤。不要一上来就设置几千并发。
- 从低并发开始,例如
Users=50,Spawn rate=5,运行3-5分钟。 - 观察响应时间(特别是95和99分位值)和失败率。如果一切正常(失败率<0.1%,P99延迟在可接受范围,如5秒内),再逐步增加并发数,例如
Users=100,150,200... 每次增加后稳定运行一段时间。 - 密切关注失败请求。一旦失败率(特别是因429限流导致的失败)显著上升,或者P99响应时间急剧增加(出现“拐点”),说明系统已经达到或接近当前配置下的极限。记录下这个临界点的并发用户数。
- 从低并发开始,例如
- 第三步:稳定性测试( soak test )。在找到的“临界点”之下,选择一个相对安全的并发数(例如临界点的80%),运行较长时间(如30分钟到1小时)。观察系统在持续负载下的表现,内存是否缓慢增长,响应时间是否保持平稳。这能发现一些在短时压力下不明显的潜在问题,如内存泄漏、连接池耗尽等。
4.2 通过命令行无头模式执行
对于自动化测试或集成到CI/CD流水线,可以使用无头(headless)模式。
# 基本命令:运行30秒,每秒启动2个用户,直到达到100个用户,然后关闭。 locust -f locustfile.py --headless --users 100 --spawn-rate 2 --run-time 30s --host https://api.example.com/v1 # 更实用的命令:指定更多参数并生成报告 locust -f locustfile.py \ --headless \ --users 500 \ # 总用户数 --spawn-rate 10 \ # 孵化率 --run-time 5m \ # 运行时间 5分钟 --csv=report \ # 生成CSV报告前缀 --csv-full-history \ # 记录整个运行过程的历史数据 --html=report.html \ # 生成HTML报告 --host=https://api.example.com/v1命令行参数解析:
--headless: 启用无头模式,不启动Web UI。--users和--spawn-rate: 等同于Web UI中的两个核心参数。--run-time: 测试运行的总时长,格式如30s,5m,1h30m。--csv和--html: 生成数据报告。CSV文件便于用Excel或Python进行后续分析,HTML报告则提供了可视化的图表和表格,非常适合存档和分享。强烈建议每次压测都生成报告。
5. 结果分析与性能瓶颈定位
压测完成后,面对Locust提供的海量数据,我们该关注什么?如何从数据中读出系统的“健康状况”和“极限所在”?
5.1 核心性能指标解读
在Locust的Web UI或生成的报告中,重点关注以下指标:
| 指标 | 含义 | 健康信号 | 报警信号 |
|---|---|---|---|
| RPS (Requests/s) | 每秒成功完成的请求数。 | 随着并发用户数增加而平稳上升。 | 达到某个点后不再增长甚至下降,说明系统吞吐量已达上限或出现瓶颈。 |
| 响应时间 (Response Times) | 请求从发出到收到完整响应所花费的时间。重点关注平均响应时间、95分位值(P95)和99分位值(P99)。 | P95/P99响应时间增长平缓,与平均响应时间差距不大。 | P95/P99响应时间急剧上升(出现“长尾”),远高于平均值。例如平均2秒,P99达到20秒,说明部分请求体验极差。 |
| 失败率 (Failures/s & Failure %) | 每秒失败的请求数及其占总请求数的百分比。 | 接近于0%(例如 < 0.1%)。 | 持续高于0.5%或突然飙升。需要立刻查看失败原因(是429限流、超时还是5xx服务器错误?)。 |
| 用户数 (Number of Users) | 当前活跃的虚拟用户数。 | 平稳达到预设的峰值并发数。 | 用户数无法达到预设值,可能因为孵化过程就出现了大量失败。 |
关键心法:寻找“拐点”。性能测试的核心不是看最高能跑到多少RPS,而是找到性能拐点。即,当并发用户数(或RPS)增加到某个值时,P99响应时间开始非线性地急剧增加,和/或失败率开始显著上升。这个点就是系统在当前配置下的有效并发处理极限。你的系统应该运行在这个拐点之前,并留有一定的安全余量(比如拐点并发数的70%)。
5.2 常见瓶颈分析与排查思路
当发现性能不佳时,如何定位问题?问题可能出在多个环节。
客户端瓶颈(你的压测机):
- 现象:Locust的CPU或内存使用率接近100%,但RPS很低。
- 排查:使用
top或htop命令查看locust进程资源消耗。单机模拟过高并发(如数千)时,可能受限于网络端口数、文件描述符数或Python GIL(对于同步客户端)。 - 解决:使用异步客户端(如我们脚本中的
aiohttp)大幅提升单机能力。如果还不够,采用Locust分布式模式,增加压测从机(worker)。
网络瓶颈:
- 现象:响应时间中“连接建立时间(TTFB前期)”占比较高,或者出现大量连接超时、重置错误。
- 排查:在压测脚本中记录更细粒度的耗时,或使用
curl -w或专业网络监控工具。检查压测机与API服务器之间的网络延迟和带宽。 - 解决:确保压测机与API服务器在同一区域网络,或使用云服务商的内网地址进行测试,以排除公网不稳定的影响。
大模型API服务端瓶颈/限流:
- 现象:这是最常见的情况。表现为大量HTTP 429(Too Many Requests)错误,或者响应时间随着并发增加呈阶梯式上升(触发了服务端的队列机制)。
- 排查:仔细查看失败请求的响应体和响应头。许多API会在响应头中返回限流信息,如
X-RateLimit-Limit,X-RateLimit-Remaining,X-RateLimit-Reset。记录并分析这些信息。 - 解决:尊重限流。根据返回的限流策略调整你的压测策略。例如,如果限流是每分钟N次请求,那么你的并发用户数和任务间隔需要据此计算。你的测试目标应该是验证在限流范围内系统的表现,而不是试图“攻破”限流。
自身应用逻辑瓶颈:
- 现象:如果你是在测试一个封装了大模型API的自家后端服务(而非直接测API),那么瓶颈可能出现在你的业务逻辑、数据库、缓存或外部依赖上。
- 排查:在Locust中为不同的请求步骤命名(如
/api/chat,/api/auth),分别观察它们的响应时间。同时,监控你的应用服务器(如Nginx, uWSGI, Gunicorn)和后端数据库的指标。 - 解决:优化慢查询,引入缓存(如Redis缓存常见问答),使用连接池管理数据库和外部API连接,考虑异步处理耗时任务等。
5.3 生成与解读HTML报告
使用--html参数生成的报告是复盘和分享的利器。报告主要包括:
- Statistics(统计概览):所有请求类型的汇总数据,包括我们最关心的RPS、响应时间分位数、失败数。
- Charts(图表):
- Total Requests per Second: RPS随时间变化曲线,看是否平稳。
- Response Times (ms): 响应时间(平均、中位数、P95)随时间变化曲线。理想状态是三条线贴近且平稳。如果P95线陡然上升,就是“拐点”的视觉体现。
- Number of Users: 并发用户数变化曲线。
- Failures(失败详情): 列出所有失败的请求,包括错误类型和发生次数,是定位问题的直接入口。
- Download Data(下载数据): 可以下载完整的CSV数据,用于更深入的自定义分析。
报告分析实战:假设你运行了一次阶梯加压测试,从报告图表中看到,当并发用户数达到180时,P95响应时间从稳定的2秒内突然跳涨到10秒以上,同时失败率开始出现429错误。那么,180就是这个API在当前Prompt长度、参数配置下的一个性能临界点。你的系统设计应该以此为依据。
6. 高级技巧与实战避坑指南
掌握了基础方法后,下面这些来自实战的经验和技巧,能让你把Locust用得更加得心应手,测试结果也更加可靠。
6.1 模拟更真实的用户行为
我们的基础脚本是每个用户随机问一个问题。真实的对话场景要复杂得多:
- 多轮对话(Session):用户会进行连续追问。这需要你在虚拟用户中维护一个对话历史(
messages列表),每次请求都将历史记录发送给API。class ConversationalUser(HttpUser): def on_start(self): self.conversation_history = [ {"role": "system", "content": "你是一个有帮助的助手。"} ] @task def multi_turn_chat(self): # 用户说 user_prompt = random.choice(PROMPT_POOL) self.conversation_history.append({"role": "user", "content": user_prompt}) # 发送整个历史 payload = { "model": MODEL_NAME, "messages": self.conversation_history, "max_tokens": 150, } # ... 发送请求 ... # 假设收到回复 # reply = data['choices'][0]['message']['content'] # 助手说 (模拟) assistant_reply = f"模拟回复 for: {user_prompt}" self.conversation_history.append({"role": "assistant", "content": assistant_reply}) # 控制对话轮次,避免历史无限增长 if len(self.conversation_history) > 10: # 保留最近10轮 self.conversation_history = [self.conversation_history[0]] + self.conversation_history[-8:] - 思考时间与行为随机性:使用
between或constant来设置wait_time只是基础。更真实的模拟可以使用constant_pacing来确保每个用户至少间隔X秒执行一次任务,或者用自定义函数实现更复杂的随机分布(如正态分布)。 - 用户分组与差异化:不是所有用户行为都一样。你可以定义多个
User类,赋予不同的weight(权重)和tasks。例如,80%的用户只进行简单问答(SimpleUser),20%的用户进行复杂多轮对话(ComplexUser)。
6.2 参数化与数据驱动
- 动态Prompt与参数:除了从文件读取Prompt池,还可以连接数据库、调用其他API动态生成测试数据。例如,测试一个摘要生成API,可以从新闻网站RSS实时获取文章标题和内容作为输入。
- 并发与Token消耗/成本估算:大模型API按Token收费。在压测脚本中,你可以解析响应,估算每次请求消耗的Prompt Token和Completion Token。结合你的目标RPS,就能粗略估算出在特定负载下每小时的API调用成本。这是一个非常重要的衍生洞察。
# 在收到响应后 if response.status == 200: data = await response.json() prompt_tokens = data.get('usage', {}).get('prompt_tokens', 0) completion_tokens = data.get('usage', {}).get('completion_tokens', 0) total_tokens = prompt_tokens + completion_tokens # 可以在这里累加统计,或者发送到外部监控系统 self.user.environment.events.request.fire( ... # 其他参数 meta={"prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens} )
6.3 我踩过的那些“坑”
- 坑:忘记设置超时。早期测试时,脚本没有设置
timeout。当API服务端偶尔无响应时,请求会一直挂起,快速耗尽压测机的可用端口和内存,导致Locust主进程崩溃。务必为所有网络请求设置合理的超时。 - 坑:API Key硬编码或误提交。曾不小心将包含测试Key的脚本提交到了GitHub公共仓库。务必使用
.env文件+python-dotenv管理密钥,并将.env加入.gitignore。 - 坑:对“成功”的定义过于简单。最初只检查HTTP状态码为200。后来发现,有些请求返回200,但响应体里是
{"error": "model overloaded"}。必须解析响应体,验证业务逻辑是否成功。 - 坑:压测环境与生产环境网络差异。在办公室网络测试一切良好,上线后用户反馈慢。原因是办公室到云服务商是优质线路,而部分用户网络质量差。压测最好在贴近生产网络环境的位置进行,比如使用与后端服务同区域的云服务器发起压测。
- 坑:忽略“预热”阶段。直接开始高并发压测,前几秒的响应时间会非常长(因为服务端冷启动、连接池建立等)。在正式记录数据前,先以一个较低的并发运行1-2分钟,让系统进入稳定状态。Locust的
--run-time包含了预热期,分析数据时可以剔除前期的异常值。
用Locust对大模型API进行并发极限评估,本质上是一个通过科学实验逼近系统真相的过程。它不能保证线上百分百不出问题,但能极大地消除不确定性,让你对系统的承载能力、响应特性和成本结构有一个量化的、坚实的理解。从编写一个模拟真实用户行为的脚本开始,到阶梯式加压观察拐点,再到深入分析性能瓶颈,每一步都需要耐心和细致。当你能够清晰地说出“我们的服务在200并发用户下,P99响应时间能稳定在3秒以内,预计每小时API成本约为X元”时,你和你的团队就拥有了做出更优技术决策和产品规划的底气。