1. 项目概述为什么选择Locust进行压测如果你正在寻找一个既能模拟海量用户又能让你用熟悉的Python代码来定义复杂用户行为的压测工具那么Locust绝对值得你花时间研究。我最初接触压测是从JMeter开始的它的图形界面和丰富的插件确实友好但当我需要模拟一些带有复杂逻辑比如根据上一个接口的返回值动态决定下一个请求的参数的业务场景时JMeter的配置就变得异常繁琐。而Locust的出现完美地解决了这个问题——它把压测脚本变成了纯粹的Python代码这意味着你可以用if-else、for循环、甚至引入外部库来处理数据灵活性直接拉满。简单来说Locust是一个开源的、分布式的负载测试工具。它的核心思想是“用代码定义用户行为”。你不需要在界面上拖拽各种元件只需要编写一个Python文件定义好User类以及它们的行为tasks然后通过命令行或Web界面启动它就能自动生成成千上万个“蝗虫”Locust意为蝗虫去“啃食”你的系统。更棒的是它自带一个实时的Web UI你可以动态调整并发用户数、观察实时的RPS每秒请求数、响应时间等关键指标测试结果还能导出进行深入分析。对于开发、测试和运维同学来说Locust提供了一条从脚本编写、测试执行到结果分析的完整链路尤其适合在敏捷开发流程中快速进行性能验证和瓶颈定位。2. 环境准备与Locust核心概念解析在开始编写第一个压测脚本之前我们需要先把舞台搭好。Locust的安装极其简单因为它是一个Python包。我强烈建议你使用虚拟环境来管理项目依赖避免污染全局的Python环境。2.1 安装与验证打开你的终端命令行执行以下命令# 使用pip安装locust pip install locust # 安装完成后验证版本 locust -V如果安装成功你会看到类似locust 2.20.0的输出。这里我推荐使用较新的2.x版本它在架构和功能上相比1.x有巨大改进比如完全基于asyncio异步IO能更高效地模拟高并发。2.2 理解Locust的脚本骨架一个最简单的Locust脚本locustfile.py看起来是这样的from locust import HttpUser, task, between class QuickstartUser(HttpUser): # 等待时间模拟用户思考时间介于1到2.5秒之间 wait_time between(1, 2.5) task def hello_world(self): # self.client是HttpUser内置的HttpSession实例用法类似requests库 self.client.get(/hello) self.client.get(/world) task(3) # 权重为3这个任务被执行的频率是hello_world任务的3倍 def view_items(self): for item_id in range(10): self.client.get(f/item?id{item_id}, name/item) self.wait_time # 每次请求后也等待一下更真实我们来拆解一下这几个核心概念HttpUser: 这是一个用户类代表一类虚拟用户。它继承自HttpUser意味着它自带了一个用于发送HTTP请求的client属性一个HttpSession对象。如果你想测试非HTTP协议如WebSocket, gRPC需要安装额外的扩展或使用User基类并自定义客户端。wait_time: 定义用户在每个任务执行后的等待时间。between(1, 2.5)使等待时间在1到2.5秒间随机分布这能更真实地模拟用户操作间隔。除了between还有constant固定等待和constant_pacing固定任务执行节奏等。task装饰器: 用来标记一个方法是用户的任务即用户会执行的操作。你可以给task传递一个整数权重权重越高该任务被选中的概率越大。上面的例子中view_items任务被执行的频率是hello_world的3倍。self.client: 这是发起HTTP请求的核心对象。它的接口设计与流行的requests库高度相似支持get、post、put、delete等方法让你几乎零成本上手。注意self.client发出的请求会自动被Locust记录统计。使用name参数可以为请求分组如上例中/item?id0到/item?id9的10个不同请求都会在统计中被归到“/item”这个名称下避免统计条目过多导致图表难以阅读。3. 编写你的第一个实战压测脚本光看骨架不够我们直接上手一个更贴近实际业务的例子模拟用户登录电商网站浏览商品列表随机查看商品详情并将随机一件商品加入购物车。3.1 定义用户行为与任务流假设我们有一个简单的电商APIPOST /api/login用于登录需要用户名和密码。GET /api/products获取商品列表。GET /api/product/{id}获取特定商品详情。POST /api/cart将商品加入购物车需要商品ID和令牌。我们的locustfile.py可以这样写from locust import HttpUser, task, between, events import random import json class EcommerceUser(HttpUser): wait_time between(2, 5) # 电商用户操作间隔稍长 host http://your-test-server.com # 被测系统的基础地址 def on_start(self): 当虚拟用户启动时执行通常用于登录等初始化操作 # 1. 登录并获取token login_payload {username: test_user, password: test_pass123} with self.client.post(/api/login, jsonlogin_payload, catch_responseTrue) as response: if response.status_code 200: response_data response.json() self.token response_data.get(token) # 假设返回中有token字段 print(fUser logged in, token: {self.token[:10]}...) else: response.failure(fLogin failed with status {response.status_code}) # 登录失败可以停止此用户的任务执行 self.stop(forceTrue) task(2) def browse_products(self): 浏览商品列表权重较高 with self.client.get(/api/products, name/api/products [列表], catch_responseTrue) as response: if response.status_code 200: products response.json() # 可以将商品列表暂存供其他任务使用这里简化处理 self.product_list products else: response.failure(fFailed to get products) task(4) def view_product_detail(self): 查看随机一个商品的详情权重最高 # 确保有商品列表 if hasattr(self, product_list) and self.product_list: product random.choice(self.product_list) product_id product[id] with self.client.get(f/api/product/{product_id}, name/api/product/[id] [详情], catch_responseTrue) as response: if response.status_code ! 200: response.failure(fFailed to get product {product_id}) else: # 如果没有商品列表先执行一次浏览商品列表任务 self.browse_products() task(1) def add_to_cart(self): 将随机一个商品加入购物车 if hasattr(self, product_list) and self.product_list and hasattr(self, token): product random.choice(self.product_list) product_id product[id] headers {Authorization: fBearer {self.token}} cart_payload {product_id: product_id, quantity: 1} with self.client.post(/api/cart, jsoncart_payload, headersheaders, name/api/cart [加入购物车], catch_responseTrue) as response: if response.status_code in [200, 201]: response.success() else: response.failure(fAdd to cart failed: {response.text})脚本解析与技巧on_start方法每个虚拟用户实例在开始执行task任务前都会先执行一次on_start。这里是执行登录、获取认证令牌等一次性初始化操作的理想位置。获取到的token存储在self.token中供后续需要认证的请求使用。catch_responseTrue参数这是Locust中非常关键的一个参数。默认情况下只要HTTP状态码不是4xx或5xxLocust就认为请求成功。但在实际业务中200状态码返回的错误信息如{code: 500, msg: 库存不足}也应被视为失败。使用catch_responseTrue后你可以通过response.failure()手动将请求标记为失败从而让统计数据更准确。任务间的数据传递注意在browse_products任务中我们将获取到的商品列表存储在self.product_list。这样在view_product_detail和add_to_cart任务中就可以随机从这个列表中选取商品。这模拟了真实用户“浏览后选择”的行为逻辑。name参数的重要性在view_product_detail的GET请求中我们使用了name/api/product/[id] [详情]。这样无论product_id是1还是100在Locust的统计图表中所有这些请求都会被聚合到“/api/product/[id] [详情]”这一项下。否则你会看到成千上万个独立的URL导致图表完全无法阅读和分析。3.2 运行压测并启动Web UI脚本写好后在终端中进入脚本所在目录运行以下命令locust -f locustfile.py默认情况下Locust的Web UI会在http://localhost:8089启动。打开浏览器访问这个地址你会看到如下启动界面Number of users (peak concurrency): 设置模拟的最大总用户数。Spawn rate (users started/second): 设置每秒启动多少用户用于控制压力爬升的斜率。Host: 被测系统的根URL如果脚本里已经写了host这里可以覆盖。填写好参数例如100个用户每秒启动10个点击“Start swarming”压测就开始了。Web UI会自动刷新展示实时数据。4. Locust Web UI 实时监控与解读Web UI是Locust的仪表盘是监控压测实时状态的第一现场。理解每个图表和数字的含义至关重要。4.1 核心监控面板详解启动压测后主界面主要分为以下几个区域Statistics统计表格 这是最重要的表格。它展示了所有请求类型的聚合性能数据。列名含义与解读Type请求名称由name参数定义。Total行是所有请求的汇总。Name同Type请求的标识名。# Requests当前已发出的总请求数。# Fails失败的请求数。务必关注此列任何非零失败都需要排查。Median (ms)中位数响应时间。50%的请求响应时间低于此值。这个值比平均响应时间更能抵抗极端值的影响更能代表“典型”用户体验。90%ile (ms)90分位响应时间。90%的请求响应时间低于此值。这是评估系统性能的关键指标它告诉你绝大多数用户的体验上限。99%ile (ms)99分位响应时间。99%的请求响应时间低于此值。用于评估极端情况下的用户体验或发现长尾问题。Average (ms)平均响应时间。容易受极值影响参考价值低于中位数和分位数。Min/Max (ms)最小/最大响应时间。偶尔看看最大值是否有异常。Average size (bytes)平均响应体大小。Current RPS当前每秒请求数。实时变化的压力指标。Current Failures/s当前每秒失败数。实操心得我通常最关注90%ile响应时间和失败率# Fails / # Requests。90%ile能告诉我大部分用户是否顺畅。如果失败率突然升高我会立即去查看“Failures”标签页和系统的错误日志。Charts图表Total Requests per Second (RPS)总RPS随时间变化曲线。理想情况下当用户数稳定后RPS也应稳定在一个区间。如果RPS持续下降可能意味着服务器处理能力达到瓶颈或正在恶化。Response Times (ms)响应时间随时间变化曲线。默认显示中位数响应时间你可以通过下拉菜单切换查看95分位、99分位等。压力上升时响应时间平缓上升是正常的但如果出现垂直飙升则很可能触发了某个性能瓶颈如数据库连接池耗尽、缓存失效等。Number of Users活跃用户数曲线。可以验证压力是否按预设的Spawn rate平稳上升并稳定在峰值。Failures列出所有失败的请求包括URL、异常信息、发生时间和所属用户。这是排查问题的第一入口。Exceptions记录脚本运行过程中抛出的Python异常比如网络错误、JSON解析错误等。Download Data可以下载CSV格式的请求统计、响应时间分布和失败记录。这是进行后续深度分析的原始数据来源务必在测试结束后下载保存。4.2 测试策略与Web UI操作技巧阶梯加压Ramp-up不要一开始就施加大压力。在“Number of users”和“Spawn rate”设置中可以先设置一个较小的目标如50用户每秒启动5个观察系统表现。稳定后再逐步提高目标值进行多轮测试。这有助于找到性能拐点。动态调整在测试运行过程中你可以随时在Web UI上修改“Number of users”并点击“Update”来动态增加或减少并发用户数。这是一个非常强大的功能可以用来做“浪涌测试”突然增加压力或“恢复测试”突然降低压力看系统恢复情况。停止与重置点击“Stop”停止生成新用户现有用户会执行完当前任务后退出。点击“New test”可以完全重置所有统计数据开始新一轮测试。5. 结果数据的深度可视化分析Web UI提供了实时视图但当我们完成一轮压测需要撰写报告或进行深度分析时就需要借助导出的CSV数据和更强大的可视化工具了。这里我主要使用Python的Pandas Matplotlib/Seaborn组合它灵活且强大。5.1 数据导出与预处理测试结束后从Web UI的“Download Data”选项卡下载三个关键CSV文件stats_requests.csv各请求类型的聚合统计类似Web UI的Statistics表格。stats_history.csv随时间变化的统计信息每秒或每若干秒一条记录包含时间戳、当前用户数、总RPS、总失败率等。failures.csv所有失败请求的详细记录。我们首先用Pandas加载和处理stats_history.csv这个文件包含了时间序列数据是绘制趋势图的基础。import pandas as pd import matplotlib.pyplot as plt import seaborn as sns # 设置中文字体和图表样式 plt.rcParams[font.sans-serif] [SimHei, Arial Unicode MS] # 根据系统选择 plt.rcParams[axes.unicode_minus] False sns.set_style(whitegrid) # 加载历史数据 df_history pd.read_csv(stats_history.csv) # 查看前几行和数据列 print(df_history.head()) print(df_history.columns) # 将时间戳列转换为datetime类型如果已经是则跳过 # 假设列名为Timestamp df_history[Timestamp] pd.to_datetime(df_history[Timestamp], units) # 如果时间戳是秒级整数 # 或者如果导出的CSV已经是可读时间字符串 # df_history[Timestamp] pd.to_datetime(df_history[Timestamp]) # 计算总失败率 df_history[Total Failure Rate] df_history[Total Failures] / df_history[Total Request Count] * 1005.2 绘制核心性能趋势图一个完整的性能分析报告需要几张关键的趋势图。fig, axes plt.subplots(3, 1, figsize(14, 10), sharexTrue) # 图1用户数与RPS趋势 ax1 axes[0] ax1.plot(df_history[Timestamp], df_history[User Count], label并发用户数, colorblue, linewidth2) ax1.set_ylabel(用户数, fontsize12) ax1.legend(locupper left) ax1_twin ax1.twinx() ax1_twin.plot(df_history[Timestamp], df_history[Total Requests/s], label总RPS, colororange, linewidth2, linestyle--) ax1_twin.set_ylabel(RPS (次/秒), fontsize12) ax1_twin.legend(locupper right) ax1.set_title(并发用户数与总RPS趋势对比, fontsize14, pad12) # 图2响应时间趋势中位数与90分位 ax2 axes[1] ax2.plot(df_history[Timestamp], df_history[Median Response Time], label中位数响应时间 (ms), linewidth2) ax2.plot(df_history[Timestamp], df_history[90%ile Response Time], label90分位响应时间 (ms), linewidth2, linestyle:) ax2.set_ylabel(响应时间 (ms), fontsize12) ax2.legend() ax2.set_title(响应时间趋势中位数 vs 90分位, fontsize14, pad12) # 图3失败率趋势 ax3 axes[2] ax3.plot(df_history[Timestamp], df_history[Total Failure Rate], label总失败率 (%), colorred, linewidth2) ax3.fill_between(df_history[Timestamp], 0, df_history[Total Failure Rate], alpha0.3, colorred) ax3.set_ylabel(失败率 (%), fontsize12) ax3.set_xlabel(时间, fontsize12) ax3.legend() ax3.set_title(总请求失败率趋势, fontsize14, pad12) # 设置一个失败率警戒线比如1% ax3.axhline(y1.0, colorgrey, linestyle--, linewidth1, alpha0.7) ax3.text(df_history[Timestamp].iloc[-1], 1.0, 1% 警戒线, verticalalignmentbottom) plt.tight_layout() plt.show()图表解读要点图1用户数 vs RPS理想情况下当用户数稳定后RPS也应稳定在一条水平线附近。如果用户数稳定但RPS持续下降说明系统吞吐量在衰减可能存在资源泄漏如内存泄漏、数据库连接未释放或外部依赖性能下降。如果RPS曲线出现剧烈锯齿状波动可能意味着网络不稳定或服务有间歇性阻塞。图2响应时间趋势重点关注90分位响应时间。当中位数还很平稳时90分位线如果开始陡峭上升往往意味着系统开始出现排队或某些慢请求拖累了整体体验。两条线的差距拉大是系统性能出现“长尾”问题的典型信号。图3失败率趋势失败率应该是0%或维持在一个极低的基线。任何持续的失败率上升都是严重警报。结合failures.csv可以定位是哪个接口在什么时间点开始大量失败。5.3 请求类型对比与分布分析除了趋势我们还需要分析不同接口的性能表现。使用stats_requests.csv。df_requests pd.read_csv(stats_requests.csv) # 过滤掉汇总行Name为“Total”和无关行 df_api df_requests[df_requests[Name] ! Total].copy() # 按请求数排序 df_api_sorted df_api.sort_values(byRequests, ascendingFalse) fig, (ax1, ax2) plt.subplots(1, 2, figsize(16, 6)) # 子图1各接口请求量分布柱状图 bars ax1.barh(df_api_sorted[Name], df_api_sorted[Requests]) ax1.set_xlabel(总请求数) ax1.set_title(各接口请求量分布) # 在柱子上标注数字 for bar in bars: width bar.get_width() ax1.text(width, bar.get_y() bar.get_height()/2, f {int(width)}, vacenter) ax1.invert_yaxis() # 让最多的在最上面 # 子图2各接口90%响应时间与失败数散点气泡图 scatter ax2.scatter( df_api[90%ile Response Time], df_api[Requests], sdf_api[Failures]*50 100, # 用失败数控制气泡大小100是基础大小 cdf_api[Median Response Time], # 用中位数响应时间着色 cmapviridis, alpha0.7, edgecolorsblack, linewidth0.5 ) ax2.set_xlabel(90分位响应时间 (ms)) ax2.set_ylabel(总请求数) ax2.set_title(接口性能气泡图大小失败数颜色中位数响应时间) # 为每个点添加标签接口名缩写 for i, row in df_api.iterrows(): ax2.annotate(row[Name].split(/)[-1][:10], # 取路径最后一部分最多10字符 (row[90%ile Response Time], row[Requests]), fontsize8, alpha0.8) # 添加颜色条 plt.colorbar(scatter, axax2, label中位数响应时间 (ms)) # 添加失败数图例示意手动创建几个示例气泡 # ... 略可根据需要添加 plt.tight_layout() plt.show()气泡图解读这张图信息量很大。X轴90分位响应时间越靠右接口越慢。Y轴总请求数越靠上该接口被调用的频率越高。气泡大小失败数气泡越大说明该接口失败越多问题可能越严重。气泡颜色中位数响应时间颜色越暖黄/白中位数响应时间越长。理想情况气泡应该集中在图表的左下角响应快、请求多且气泡很小失败少、颜色偏冷中位数也快。问题接口位于右上角的大气泡这是最危险的接口响应慢、失败多是首要优化目标。位于右侧的小气泡虽然调用不多、失败也少但响应很慢。可能需要检查是否是低频但复杂的查询或者存在优化空间。位于中部或左侧的大气泡调用频繁失败数多但响应可能还行。这通常意味着接口的异常处理或边界条件有问题成功率低。6. 分布式压测与高级配置当单台压测机无法产生足够压力或者想避免压测机自身成为瓶颈时就需要进行分布式压测。Locust的分布式模式采用主从master-worker架构。6.1 启动分布式集群假设你有三台机器一台作为主节点Master两台作为工作节点Worker。在主节点上启动Master# 在主节点机器上 locust -f locustfile.py --master --hosthttp://your-test-server.comMaster节点本身不模拟用户只负责协调、收集数据和提供Web UI。在每个工作节点上启动Worker# 在工作节点1和2上分别执行将 --master-host 指向主节点的IP locust -f locustfile.py --worker --master-host192.168.1.100你需要将192.168.1.100替换为你主节点的实际IP地址。启动后在Master的Web UI上你应该能看到“Workers”区域显示已连接的Worker数量例如2/2。此时你在Web UI上设置的用户数将会由Master动态分配给两个Worker共同承担。6.2 分布式压测的注意事项与技巧数据一致性确保所有Worker节点上的locustfile.py和相关的测试数据如CSV账户文件是完全相同的。避免共享状态在User类中避免使用类变量class variable来共享状态因为不同的Worker进程不共享内存。每个User实例应该是独立的。如果需要在用户间共享只读数据如商品列表可以考虑在on_start中从公共数据源如文件、Redis读取。主节点网络确保Worker节点能通过网络访问到Master节点默认端口5557用于Worker连接8089用于Web UI。压力机资源监控Worker节点本身的CPU、内存和网络带宽。如果Worker节点资源吃满它本身就会成为瓶颈产生不了足够压力。此时需要增加Worker节点。使用--expect-workers启动Master时可以使用--expect-workers 4参数指定期望连接的Worker数量。只有当指定数量的Worker全部连接后Web UI上才会出现“Start”按钮这可以防止Worker未就绪就启动测试。6.3 使用FastHttpUser提升性能Locust默认的HttpUser使用requests库它是同步的。对于极高的并发如数万用户requests库和Python的线程/进程模型可能成为瓶颈。此时可以切换到FastHttpUser它使用gevent和httptools性能更高。from locust import task, between from locust.contrib.fasthttp import FastHttpUser # 注意导入路径 class FastUser(FastHttpUser): wait_time between(1, 3) task def index(self): with self.client.get(/, catch_responseTrue) as resp: if resp.status_code 200: resp.success()切换后你可能会发现单机所能模拟的用户数大幅提升。但请注意FastHttpUser的API与HttpUser的client略有不同需要查阅对应文档。7. 常见问题排查与实战心得压测过程中总会遇到各种问题这里记录几个我踩过的坑和解决方法。7.1 “Socket”相关错误问题大量报错ConnectionResetError: [Errno 104] Connection reset by peer或ConnectionRefusedError。排查检查被测服务首先确认被测服务是否存活日志是否有大量错误如Too many open files。检查压测机限制Linux系统下单个进程能打开的文件描述符数量有限。使用ulimit -n查看。如果模拟的用户数很大每个并发连接都可能占用一个文件描述符很容易超限。检查端口范围操作系统可用的临时端口范围有限。使用sysctl net.ipv4.ip_local_port_range查看。当压测机作为客户端需要建立大量到同一服务器端口的连接时可能会耗尽临时端口。解决# 临时提高压测机的文件描述符限制 ulimit -n 65535 # 临时扩大压测机的本地端口范围 sudo sysctl -w net.ipv4.ip_local_port_range1024 65535更佳做法是将这些配置写入压测机的/etc/security/limits.conf和/etc/sysctl.conf文件并重启生效。7.2 响应时间异常飙升但CPU/内存不高问题在Web UI上看到响应时间曲线突然变成一条垂直的线但登录服务器查看CPU和内存使用率并不高。排查检查外部依赖你的应用可能依赖数据库、缓存Redis、消息队列Kafka或第三方API。瓶颈很可能出现在这些地方。检查这些中间件的监控连接数、QPS、慢查询。检查应用线程池/连接池这是最常见的原因之一。例如数据库连接池maxActive设置过小或者应用服务器如Tomcat的线程池满了请求都在排队等待获取资源。检查锁竞争应用内部是否存在激烈的锁竞争如synchronized、ReentrantLock可以使用arthas、async-profiler等工具进行现场诊断。解决根据排查结果调整对应资源池的配置。对于数据库可能需要优化慢SQL或增加连接池大小。对于应用可能需要调整线程池配置或优化锁粒度。7.3 Locust Worker节点失去连接问题在分布式压测中Worker节点突然从Master的Web UI上消失或者控制台报错断开连接。排查网络问题Master和Worker之间网络不稳定。Worker进程崩溃检查Worker节点的日志看是否有Python异常导致进程退出。消息队列积压如果压测产生的数据量非常大比如每个请求的响应体都很大Master和Worker之间的消息队列可能积压导致心跳超时。解决确保网络稳定。在Worker启动命令中添加--logfile worker.log将日志输出到文件便于排查。可以尝试增加Master和Worker之间的心跳超时时间通过环境变量LOCUST_EXPECT_WORKERS_WAIT单位秒但这通常是治标不治本根本还是要优化脚本或增加Master节点资源。7.4 个人实战心得脚本设计要贴近真实用户不要只发请求。加入合理的wait_time设计有逻辑依赖的任务流如先登录再操作使用变量参数化请求如不同的用户ID、商品ID。这样的测试结果更有参考价值。始终关注失败率在压测过程中我习惯把Web UI的“Failures”标签页一直开着。一旦出现失败立即暂停增压分析原因。带着错误继续压测得到的数据是无效的。性能测试要有基准在每次代码发布或配置变更前跑一套固定的性能测试用例保存关键指标如90%ile响应时间、最大稳定RPS。这样就能清晰地看出这次变更对性能的影响是正面的还是负面的。结果分析要结合系统监控压测工具的数据响应时间、RPS只是表象。一定要结合被测系统的监控图表CPU、内存、磁盘IO、网络流量、JVM GC、数据库监控等一起看。当响应时间变慢时去系统监控里找对应的资源瓶颈点这样才能定位到根本原因。循序渐进持续学习从单接口压测开始再到场景压测最后是全链路压测。Locust只是一个工具真正的挑战在于如何设计测试场景、如何分析瓶颈、如何与开发团队协作优化。每次压测都是一次学习系统架构的机会。
