1. 项目概述为什么是Locust如果你做过性能测试肯定对JMeter、LoadRunner这些名字不陌生。它们功能强大但有时候也让人觉得“笨重”——图形界面操作、复杂的线程组配置、报告生成后还得花时间分析。特别是当你需要快速验证一个API接口的吞吐量或者想用代码更灵活地模拟复杂用户行为时这些工具的学习成本和灵活性就成了瓶颈。这就是Locust出场的时候了。Locust是一个用Python写的开源负载测试工具。它的核心哲学是“用代码定义用户行为”。这意味着你不再需要点来点去配置虚拟用户而是像写普通的Python脚本一样描述你的用户会做什么先访问首页登录然后搜索商品最后下单。整个过程清晰、可版本控制并且能利用Python生态的所有库比如requests, httpx, websockets来构建更真实的测试场景。我最初接触Locust是因为一个电商促销活动的压力测试。当时需要模拟数万用户在不同时间段、以不同概率执行浏览、加购、秒杀等行为。用JMeter实现这种带有条件逻辑和状态保持的流程非常痛苦而用Locust我只需要几十行Python代码就搞定了还能实时在Web界面上看到RPS每秒请求数和响应时间的变化那种掌控感是传统工具很难给的。所以这篇实战教程的目标很明确带你从零开始掌握用Locust构建可编程、高性能负载测试的能力。无论你是想测试一个简单的REST API还是一个包含WebSocket的实时应用甚至是需要处理自定义协议的系统Locust都能提供一套简洁而强大的框架。我们不止讲怎么用更会深入为什么这么用以及在实际项目中踩过的那些坑。2. 环境准备与Locust核心概念解析工欲善其事必先利其器。在开始写第一个Locust脚本之前我们需要把环境搭好并理解它的几个核心运行模式。2.1 安装与快速验证Locust的安装极其简单因为它就是一个Python包。我强烈建议使用虚拟环境来管理依赖避免污染系统环境。# 1. 创建并进入一个项目目录 mkdir locust_demo cd locust_demo # 2. 创建虚拟环境以venv为例conda或pipenv同理 python -m venv venv # 3. 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 4. 安装locust pip install locust安装完成后可以通过locust -V命令验证安装是否成功。接下来我们创建一个最简单的测试脚本命名为locustfile.py。这个文件名是Locust默认寻找的入口文件。# locustfile.py from locust import HttpUser, task, between class QuickstartUser(HttpUser): # 模拟用户在每个任务执行后等待1到5秒 wait_time between(1, 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): # 注意这里使用了name参数来分组请求否则报告中会显示10个不同的URL self.client.get(f/item?id{item_id}, name/item)这个脚本定义了一类用户QuickstartUser他们会随机等待1-5秒然后执行任务。task装饰器定义了一个任务数字权重表示执行频率的相对比例。self.client.get会发送HTTP GET请求。现在在终端运行Locustlocust默认会启动一个Web UI在http://localhost:8089。打开浏览器输入目标主机的URL例如http://your-test-server.com设置要模拟的用户总数和每秒启动的用户数就可以开始测试了。注意很多新手会在这里困惑locustfile.py里写的URL是路径如/hello在Web UI里填的Host才是完整的主机部分。两者拼接起来才是最终的请求地址。例如脚本里写self.client.get(“/api/login”)UI里Host填https://api.example.com那么实际请求的就是https://api.example.com/api/login。2.2 核心运行模式单机与分布式Locust主要有两种运行模式理解它们对后续测试至关重要。1. 单机模式就是上面演示的直接运行locust命令。所有的虚拟用户Worker和Web UIMaster都运行在同一个进程里。这适合快速验证和小规模测试通常几千用户以内。它的优点是简单开箱即用。2. 分布式模式当需要模拟数万甚至百万级用户时单台机器的网络和CPU可能成为瓶颈。Locust的分布式模式采用Master-Worker架构。Master节点负责分发测试任务、收集汇总来自所有Worker的测试数据并提供Web UI。它本身不模拟用户。Worker节点负责真正地执行测试脚本生成负载。可以启动多个Worker进程甚至分布在多台物理机器上。启动分布式模式的命令如下# 在Master机器上启动 locust -f locustfile.py --master --hosthttp://your-test-server.com # 在每台Worker机器上启动 locust -f locustfile.py --worker --master-hostMASTER_IP这里有个关键点你的测试脚本locustfile.py必须在Master和所有Worker节点上都存在并且内容一致。通常的做法是将脚本放在版本控制系统如Git中在所有测试机器上拉取。实操心得在分布式测试时确保Worker节点与Master节点之间的网络通畅且防火墙开放了Locust默认的通信端口5557用于Worker注册5558用于数据通信。我曾遇到过因为Worker节点时间不同步导致测试数据时间戳混乱的问题所以最好在所有节点上配置NTP时间同步。2.3 理解User、Task与Events这是Locust脚本编程模型的三个核心概念。User类代表一类虚拟用户的行为模式。你通过继承HttpUser用于HTTP测试或User用于非HTTP协议来定义。每个虚拟用户实例在其生命周期内会独立运行你定义的任务。Task任务是用户具体执行的操作。使用task装饰器定义。Locust会按照权重随机选择任务执行。你还可以通过on_start和on_stop方法定义用户开始和结束时的动作如登录和注销。Events事件钩子。这是Locust非常强大的扩展机制。它允许你在测试生命周期的特定时刻注入自定义逻辑。例如test_start/test_stop: 整个测试开始和结束时触发。init 每个Locust进程初始化时触发常用于设置全局变量。request 每次请求发送前后触发可以用于自定义统计或修改请求。下面是一个使用事件钩子初始化全局数据库连接池的例子from locust import events from locust.runners import MasterRunner import redis redis_pool None events.init.add_listener def on_locust_init(environment, **kwargs): # 仅在Master节点或单机模式初始化时执行 if not isinstance(environment.runner, MasterRunner): global redis_pool redis_pool redis.ConnectionPool(hostlocalhost, port6379, decode_responsesTrue) print(Redis连接池初始化完成) events.test_stop.add_listener def on_locust_stop(environment, **kwargs): if redis_pool: redis_pool.disconnect() print(Redis连接池已关闭)通过事件机制你可以让Locust与你的监控系统、数据生成器、外部服务等深度集成。3. 构建复杂的用户行为模拟真实的用户行为很少是简单、随机的请求。他们会有序列、有状态、有思考时间。Locust的编程模型让模拟这些行为变得非常直观。3.1 任务序列与权重控制最简单的任务定义是使用task装饰器。Locust默认会根据任务权重进行随机选择。但有时我们需要模拟一个固定的业务流程比如“必须先登录才能下单”。这时可以使用SequentialTaskSet。from locust import HttpUser, task, SequentialTaskSet, between class UserBehaviour(SequentialTaskSet): # 这个类里的任务会按定义顺序执行 task def login(self): resp self.client.post(/login, json{username: test, password: test}) if resp.status_code 200: self.token resp.json().get(token) # 将token保存为用户实例的状态 task def browse_product(self): # 可以使用上一步保存的状态 headers {Authorization: fBearer {self.token}} if hasattr(self, token) else {} self.client.get(/products, headersheaders) task def stop(self): # SequentialTaskSet需要一个任务来中断序列否则会循环执行 self.interrupt() class WebsiteUser(HttpUser): tasks [UserBehaviour] # 将TaskSet类作为任务列表 wait_time between(2, 5)SequentialTaskSet会按顺序执行其内部的任务。注意最后一个stop任务调用了self.interrupt()这会让用户跳出这个序列然后根据wait_time等待后再次从UserBehaviour的第一个任务开始。这样就模拟了用户“登录-浏览-等待-再登录-浏览”的循环。对于更复杂的权重控制你可以直接给tasks属性赋一个列表列表元素可以是任务方法也可以是(callable, weight)元组。class WebsiteUser(HttpUser): wait_time between(1, 3) def login(self): # ... 登录逻辑 def logout(self): # ... 登出逻辑 def view_index(self): # ... 浏览首页 # 定义任务列表view_index执行概率是login的5倍是logout的10倍 tasks [ view_index, # 权重默认为1 (login, 2), # 权重为2 (logout, 1) # 权重为1 ]3.2 参数化与测试数据管理压测时使用固定的测试数据如固定的用户名、商品ID会导致缓存命中率异常高无法反映真实场景。我们需要参数化数据。1. 队列循环取用适用于需要保证数据唯一性如注册新用户的场景。from locust import HttpUser, task, between import queue class ApiUser(HttpUser): wait_time between(0.5, 2) # 在类级别初始化一个队列 user_credentials queue.Queue() # 假设我们准备1000个测试账号 for i in range(1000): user_credentials.put({username: ftest_user_{i}, password: 123456}) def on_start(self): # 每个虚拟用户启动时从队列中取出一组凭证 try: self.credential self.user_credentials.get_nowait() except queue.Empty: # 如果数据用完可以停止用户或者抛出异常结束测试 print(测试数据已耗尽) self.stop(forceTrue) task def login(self): # 使用分配到的凭证登录 resp self.client.post(/api/login, jsonself.credential) if resp.status_code ! 200: # 登录失败可以将凭证放回队列重试或者记录错误 print(f登录失败: {self.credential[username]})2. 随机选取适用于浏览、查询等不需要严格唯一性的场景。import random from locust import HttpUser, task, between class SearchUser(HttpUser): wait_time between(1, 3) # 准备一个关键词列表 search_keywords [手机, 笔记本电脑, 耳机, 书籍, 运动鞋, 咖啡, 旅游, 编程教程] task def search(self): keyword random.choice(self.search_keywords) # 使用name参数聚合相同模式的请求否则报告中会出现大量不同的URL self.client.get(f/search?q{keyword}, name/search?q[keyword])3. 从外部文件读取当数据量很大时如10万个商品ID更适合从CSV或JSON文件读取。import csv from locust import HttpUser, task, between class ProductUser(HttpUser): wait_time between(0.1, 0.5) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 注意每个用户实例化时都会读取文件在生产中应优化为全局共享 self.product_ids [] with open(product_ids.csv, r) as f: reader csv.reader(f) for row in reader: self.product_ids.append(row[0]) task def view_product(self): if self.product_ids: pid random.choice(self.product_ids) self.client.get(f/product/{pid}, name/product/[id])注意事项从文件读取数据要特别注意性能。如果文件很大在每个用户实例化时都读取会消耗大量内存和IO。更好的做法是利用events.init事件在测试初始化时一次性将数据加载到全局变量或共享内存中供所有Worker进程内的用户实例访问。同时要考虑多进程下的数据同步问题可以使用queue.Queue或multiprocessing.Manager来管理共享数据。3.3 处理关联与状态保持现代Web应用大量使用Cookie、Session和Token来保持用户状态。Locust的HttpUser内置的client属性一个requests.Session的子类会自动处理Cookie。对于Token我们需要手动管理并将其添加到后续请求的头部。class ApiUser(HttpUser): wait_time between(1, 3) host https://api.example.com def on_start(self): # 用户启动时登录并获取token login_resp self.client.post(/v1/auth/login, json{user: test, pass: test}) if login_resp.ok: self.auth_token login_resp.json()[access_token] # 后续请求的client会自动使用同一个Session但不会自动加Token else: self.stop(forceTrue) # 登录失败则停止该用户 task def get_profile(self): # 手动设置认证头 headers {Authorization: fBearer {self.auth_token}} self.client.get(/v1/user/profile, headersheaders) task def create_order(self): headers {Authorization: fBearer {self.auth_token}} self.client.post(/v1/orders, json{item: abc}, headersheaders)对于更复杂的场景比如一个请求的返回值是下一个请求的参数典型的如先创建订单获得订单号再支付你需要解析响应并传递状态。task def create_and_pay_order(self): # 1. 创建订单 create_resp self.client.post(/api/order, json{items: [{id: 1}]}) if create_resp.ok: order_id create_resp.json()[orderId] # 2. 使用上一步的orderId进行支付 pay_resp self.client.post(f/api/order/{order_id}/pay, json{method: credit_card}) # 可以根据pay_resp的结果决定后续任务分支 if not pay_resp.ok: self.client.post(f/api/order/{order_id}/cancel) # 支付失败则取消订单这种链式调用能非常真实地模拟用户的连续操作流程。4. 高级配置、监控与结果分析一个专业的性能测试不仅仅是把请求发出去还要能精细地控制负载模型、实时监控关键指标并能对测试结果进行深入分析。4.1 负载模型配置与自定义客户端Locust的负载模型主要由wait_time控制。除了between还有constant(n) 每次任务后固定等待n秒。constant_pacing(n) 尝试让每次任务循环执行等待正好花费n秒。如果任务执行超过n秒则等待时间为0。这常用于模拟固定的节奏。from locust import HttpUser, task, constant_pacing class FixedPacingUser(HttpUser): wait_time constant_pacing(2) # 目标每2秒完成一个任务循环包括执行和等待 task def my_task(self): self.client.get(/api)有时你需要测试非HTTP协议的服务比如WebSocket、gRPC、自定义TCP协议等。这时你需要继承基础的User类并自定义客户端。from locust import User, task, between import websocket import json class WebSocketClient: def __init__(self, host): # 建立WebSocket连接 self.ws websocket.create_connection(fws://{host}/chat) def send(self, message): # 发送消息并记录响应时间模拟Locust的请求统计 start_time time.time() self.ws.send(json.dumps(message)) response self.ws.recv() # 阻塞接收 total_time int((time.time() - start_time) * 1000) # 毫秒 # 这里可以手动触发Locust的请求事件以便在统计中记录 from locust import events events.request.fire( request_typeWS, namechat_send, response_timetotal_time, response_lengthlen(response), exceptionNone, ) return json.loads(response) def close(self): self.ws.close() class WebSocketUser(User): wait_time between(1, 3) def on_start(self): self.client WebSocketClient(self.host) # self.host是运行Locust时指定的--host def on_stop(self): self.client.close() task def send_message(self): self.client.send({type: msg, content: Hello Locust})通过自定义客户端和手动触发events.request事件你可以将任何协议的测试集成到Locust的统计和报告中。4.2 实时监控与自定义指标Locust的Web UI提供了基本的监控图表RPS、响应时间、用户数。但对于企业级测试我们往往需要更细致的监控比如业务关键接口的成功率。特定步骤的响应时间百分位数如P99。与系统监控如服务器CPU、内存、数据库连接数关联。方法一使用Locust的事件钩子记录自定义数据。from locust import events from locust.runners import STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP import time # 存储自定义统计 custom_stats {} events.request.add_listener def on_request(request_type, name, response_time, response_length, exception, context, **kwargs): # 每次请求完成后触发 if name not in custom_stats: custom_stats[name] {count: 0, total_time: 0, errors: 0} custom_stats[name][count] 1 custom_stats[name][total_time] response_time if exception: custom_stats[name][errors] 1 events.test_stop.add_listener def on_test_stop(environment, **kwargs): # 测试结束时打印自定义报告 print(\n 自定义请求统计 ) for name, stats in custom_stats.items(): if stats[count] 0: avg_time stats[total_time] / stats[count] error_rate (stats[errors] / stats[count]) * 100 print(f{name}: 请求数{stats[count]}, 平均响应时间{avg_time:.1f}ms, 错误率{error_rate:.2f}%)方法二将数据推送到外部监控系统如Prometheus、InfluxDB。from locust import events from prometheus_client import Counter, Histogram, push_to_gateway import time # 定义Prometheus指标 REQUEST_COUNT Counter(locust_requests_total, Total requests, [method, endpoint, status]) REQUEST_LATENCY Histogram(locust_request_latency_seconds, Request latency, [method, endpoint]) events.request.add_listener def track_request(request_type, name, response_time, response_length, exception, **kwargs): status failure if exception else success REQUEST_COUNT.labels(methodrequest_type, endpointname, statusstatus).inc() REQUEST_LATENCY.labels(methodrequest_type, endpointname).observe(response_time / 1000.0) # 秒 events.test_stop.add_listener def push_metrics(environment, **kwargs): # 测试结束时将指标推送到PushGateway from prometheus_client import pushadd_to_gateway pushadd_to_gateway(localhost:9091, joblocust_test, registryREQUEST_COUNT._collector())4.3 测试结果分析与报告生成Locust运行结束后可以在Web UI上点击“Download Data”下载CSV报告包含请求和分布时间的详细数据。但对于自动化测试和持续集成我们需要命令行工具和更程序化的方式。1. 无头模式运行与结果导出使用--headless模式可以在不启动Web UI的情况下运行测试并通过--csv参数导出结果。locust -f locustfile.py --headless --hosthttp://your-server.com \ --users 100 --spawn-rate 10 --run-time 5m \ --csvresults/my_test这条命令会以每秒10个用户的速度启动直到达到100个用户然后运行5分钟。结束后会生成my_test_requests.csv和my_test_stats.csv等文件。2. 深入分析CSV报告_stats.csv文件包含了每个请求的聚合数据。你可以用Pandas或Excel进行深入分析import pandas as pd df pd.read_csv(my_test_stats.csv) # 计算总体成功率 total_requests df[Request Count].sum() failed_requests df[Failure Count].sum() success_rate (1 - failed_requests/total_requests) * 100 print(f总请求数: {total_requests}, 成功率: {success_rate:.2f}%) # 找出响应时间最长的接口 slowest df.nlargest(5, Average Response Time)[[Name, Average Response Time, # requests]] print(响应时间最长的接口) print(slowest) # 分析响应时间分布需要 _distribution.csv 文件 dist_df pd.read_csv(my_test_distribution.csv) # 可以绘制某个接口的响应时间百分位数图表3. 识别性能瓶颈分析报告时不要只看平均响应时间。关注百分位数如P95, P99和错误率。P99响应时间激增可能意味着系统在某些边缘情况下如缓存失效、数据库锁性能骤降。特定接口错误率飙升可能是该接口存在资源竞争、连接池耗尽或代码bug。随着用户数增加RPS增长平缓甚至下降说明系统已经达到瓶颈可能是CPU、内存、网络带宽或外部依赖如数据库达到了极限。结合系统监控如服务器的CPU、内存、磁盘IO、网络流量数据库的QPS、连接数、慢查询可以更准确地定位瓶颈所在。例如如果RPS上不去但CPU使用率很低那么瓶颈可能在于数据库响应慢或应用代码中有同步阻塞操作。5. 实战中的常见问题与排查技巧在实际项目中使用Locust你肯定会遇到各种意想不到的情况。下面是我总结的一些典型问题和解决方法。5.1 性能瓶颈不在被测系统而在Locust本身现象增加虚拟用户数后Locust Worker节点的CPU或内存占用率很高但被测系统的压力并没有成比例增加甚至RPS开始下降。原因与解决脚本逻辑过于复杂如果在task方法中执行了大量的CPU密集型计算如加解密、复杂字符串处理会消耗Locust进程自身的资源。技巧将数据预处理工作移到on_test_start事件中或者使用更高效的数据结构。避免在任务循环内进行繁重计算。单个locustfile.py文件过大当定义的用户类和任务非常多时Python导入和运行开销会增大。技巧合理拆分代码。可以使用模块化将不同的用户行为定义在不同的Python文件中然后通过--locustfile指定主文件或在主文件中导入。网络连接未复用虽然HttpUser的client默认使用Session但如果你在每个任务里都创建新的requests会话或没有正确关闭连接会导致端口耗尽和额外开销。技巧坚持使用self.client发起请求。对于自定义协议客户端确保连接在on_start创建在on_stop关闭。单机资源限制模拟的用户数太多单台机器无法承载。解决使用分布式模式。将Worker部署到多台机器上。这是解决此问题的根本方法。记得监控Worker机器本身的资源使用情况。5.2 “Socket” 或 “Connection Reset” 错误激增现象测试中后期出现大量连接错误如ConnectionResetError,TimeoutError。排查步骤检查被测服务首先登录被测服务器查看系统日志、应用日志确认是否服务本身崩溃、重启或达到了最大连接数限制如netstat -an | grep :80 | wc -l。检查Locust端操作系统限制Linux系统下每个进程能打开的文件描述符数量有限。当模拟数万并发连接时可能触顶。解决使用ulimit -n 65535命令提高限制或修改/etc/security/limits.conf文件永久生效。TCP端口耗尽Locust作为客户端每个TCP连接需要一个本地端口。短时间内产生大量短连接可能导致本地端口被占满需要等待TIME_WAIT状态结束。解决启用连接复用确保使用HttpUser的client它底层是requests.Session默认启用HTTP Keep-Alive。调整内核参数Linux# 减少TIME_WAIT等待时间 sysctl -w net.ipv4.tcp_fin_timeout30 # 允许端口快速重用 sysctl -w net.ipv4.tcp_tw_reuse1 sysctl -w net.ipv4.tcp_tw_recycle1 # 注意在NAT环境下慎用此参数 # 增加本地端口范围 sysctl -w net.ipv4.ip_local_port_range1024 655355.3 测试结果波动大无法复现现象同样的脚本、同样的配置两次测试的结果如平均响应时间、RPS差异很大。可能原因测试环境不干净被测系统存在其他干扰流量数据库缓存未预热测试数据每次不同。解决确保测试环境独立。正式压测前先进行一段时间的“预热”运行让系统缓存如数据库查询缓存、JVM JIT编译稳定下来。使用固定或可重复生成的测试数据集。负载模型不合理使用完全随机的wait_time可能导致请求分布不均匀。解决尝试使用constant_pacing来产生更稳定、可预测的负载。或者根据业务日志分析真实的用户思考时间分布用更复杂的自定义wait_time类来模拟。外部依赖波动如果被测系统依赖第三方API、数据库集群或其他微服务这些依赖的性能波动会直接影响结果。解决在测试期间同时监控这些外部依赖的性能指标。如果可能在测试环境中Mock掉不稳定的外部依赖或者使用其测试环境。5.4 如何测试需要图形验证码或复杂前端交互的接口这是一个常见挑战因为Locust本质是协议级测试工具不执行JavaScript。策略绕过与开发团队协商在测试环境中禁用验证码或提供万能验证码如输入任意6位数字即可通过。这是最直接有效的方法。Mock如果验证码是服务端生成的可以让开发提供一个获取验证码的测试接口或者直接让登录接口接受一个特定的“测试Token”来跳过验证。OCR不推荐用于大规模压测对于必须破解验证码的场景可以使用Tesseract等OCR库。但这会极大增加单个请求的处理时间严重影响Locust的发压能力只适用于小规模功能性验证。import pytesseract from PIL import Image import io task def login_with_captcha(self): # 1. 获取验证码图片 captcha_resp self.client.get(/captcha) image Image.open(io.BytesIO(captcha_resp.content)) # 2. 识别耗时操作 captcha_text pytesseract.image_to_string(image).strip() # 3. 使用识别结果登录 self.client.post(/login, data{user: test, captcha: captcha_text})务必注意这种方法会严重拖慢单个虚拟用户的执行速度你需要启动更多的用户才能产生同样的压力极大地增加了负载生成器的资源消耗。5.5 分布式测试时数据不同步或汇总异常现象在Master的Web UI上看到的总RPS低于各Worker节点之和或者某些请求的统计缺失。排查检查网络和防火墙确保所有Worker节点能通过TCP端口5557和5558连接到Master节点。使用telnet master_ip 5557命令测试。检查系统时间确保Master和所有Worker节点的系统时间基本同步差异在秒级以内。时间不同步会导致数据的时间戳混乱影响聚合。检查脚本一致性确保所有节点上的locustfile.py完全一致特别是请求的name参数。如果不同Worker对同一URL使用了不同的name在Master的统计中它们会被视为不同的请求。Master节点资源不足当Worker数量很多比如上百个时Master节点汇总数据可能需要大量CPU和内存。确保Master节点配置足够。最后性能测试本身是一个“测不准”与“不断逼近真相”的过程。Locust给了你一把强大且灵活的武器但如何设计测试场景、如何分析结果、如何定位瓶颈更需要你对业务系统架构的深入理解。每一次压测都是一次对系统认知的深化。不要只满足于跑出一个数字更要理解这个数字背后的意义以及当这个数字发生变化时系统究竟发生了什么。