1. 项目概述为什么是Locust如果你是一名开发者或者负责过线上服务的稳定性保障那你一定对“性能测试”这个词不陌生。当你的应用上线前你肯定想知道它能扛住多少用户同时访问响应时间会不会随着用户增多而急剧变慢服务器资源会不会被瞬间打满这些问题靠拍脑袋是没用的必须靠实实在在的压测数据来说话。市面上性能测试工具不少老牌的JMeter功能强大但配置繁琐写个脚本像在画流程图LoadRunner更是重量级选手价格不菲。对于很多以Python技术栈为主的团队或者希望测试脚本能更灵活、更“程序员友好”的开发者来说Locust的出现就像一股清流。Locust是一个用Python写的开源负载测试工具。它的核心思想非常极客你用Python代码来定义用户行为。你想模拟用户登录、浏览商品、下单支付没问题就像写普通的业务逻辑代码一样去写。这种“代码即脚本”的方式让测试逻辑的表达能力直接拉满也让它天然地融入了CI/CD流程。我第一次接触Locust是因为一个紧急的促销活动压测需求当时用JMeter临时拼凑脚本被那些XML配置和元件搞得头大。换成Locust后一个下午就写出了覆盖全链路的压测脚本那种“用代码掌控一切”的感觉瞬间就爱上了。简单来说Locust适合你如果你1熟悉Python2厌倦了图形界面工具的笨重和限制3希望压测脚本能像项目代码一样被版本管理、被复用、被集成4需要快速验证API或Web服务的性能瓶颈。2. Locust的核心架构与工作原理拆解要玩转一个工具先得理解它的大脑和四肢。Locust的架构设计得很清晰理解了它你写脚本和排查问题都会事半功倍。2.1 核心组件蝗虫、蜂巢与王Locust的命名很有意思直译就是“蝗虫”。在它的世界里有三个核心角色User类蝗虫这是你脚本的核心。你需要定义一个继承自HttpUser或更基础的User的类。这个类的每一个实例在压测运行时都代表一个并发用户。你在类里面定义这个用户会做什么比如先访问首页然后登录再查询数据。TaskSet任务集你可以把它理解为用户行为的“场景剧本”。一个User可以包含多个TaskSet用来组织复杂的、有顺序或权重的用户操作。比如你可以定义一个“浏览任务集”只包含查看商品列表、商品详情和一个“购买任务集”包含加购、下单、支付。TaskSet让代码结构更清晰。Master主节点与 Worker从节点这是Locust支持分布式压测的关键。当你需要模拟成千上万的用户时单机可能无法生成足够的压力或者本身成为瓶颈。这时你可以启动一个Master节点和多个Worker节点。Master节点负责协调它不模拟任何用户只负责分发测试任务、收集所有Worker节点的统计数据并托管Web UI那个实时展示图表的管理界面。Worker节点干活的“苦力”。它们接收Master的指令真正地启动和运行你定义的User类实例模拟用户行为并向Master报告自己的数据。这种主从架构让你可以轻松地利用多台机器的资源发起大规模的压力测试。比如用一台配置一般的机器做Master用几台高配的云服务器做Worker就能轻松模拟出数万甚至数十万的并发用户。2.2 工作流程从代码到压力当你运行一个Locust脚本时背后发生了什么脚本加载Locust启动加载你的Python脚本识别出其中定义的User类。用户孵化根据你在Web UI或命令行中设定的用户数--users和孵化速率--spawn-rateLocust会动态地创建User类的实例。每个实例都是一个独立的绿色线程gevent协程而不是操作系统线程因此资源开销极小单机也能模拟数千并发。任务执行每个用户实例协程会独立地、循环地执行你定义的任务tasks。Locust会按照你给任务设置的权重task(3)随机挑选任务执行并在每个任务之间插入一个随机的等待时间通过wait_time属性设置如between(1, 5)表示等待1到5秒以此来模拟真实用户思考、点击的间隔避免产生完全不间断的“机枪”式请求那不符合真实场景。请求发送与统计当执行到具体的HTTP请求调用如self.client.get(/api/data)时Locust会记录下这个请求的发送时间。收到响应后再记录下响应时间、状态码、响应大小等。这些数据会实时汇总。数据聚合与展示所有数据被汇聚到Master单机运行时就是本机进程Locust实时计算并展示在Web UI上包括每秒请求数RPS、响应时间平均、中位数、P95、P99、当前在线用户数等关键指标。注意这里有一个非常重要的细节。Locust的“并发用户数”和你常听到的“每秒请求数RPS”是两个概念。1000个并发用户如果每个用户平均每5秒发一个请求那么RPS大概在200左右。Locust控制的是并发用户的数量和他们的行为模式最终的RPS是结果而不是直接设定的参数。这更符合真实用户场景。3. 从零开始编写你的第一个Locust压测脚本理论说再多不如动手写一行代码。我们来创建一个最经典的场景压测一个简单的用户查询API。3.1 环境准备与安装首先确保你有一个Python环境3.6及以上。使用pip安装Locust非常简单pip install locust安装完成后在命令行输入locust --help如果能显示帮助信息说明安装成功。3.2 脚本结构解剖创建一个名为locustfile.py的文件这是Locust默认寻找的脚本文件名。我们来逐部分解析一个基础脚本from locust import HttpUser, task, between class QuickstartUser(HttpUser): 定义一个模拟用户类继承自HttpUser。 这个类的每个实例在压测中都是一个独立的并发用户。 # 设置用户在每个任务执行后的等待时间范围单位秒 # 这用于模拟真实用户操作之间的间隔避免产生不合理的持续高压力 wait_time between(1, 3) # 使用task装饰器来标记这是一个用户任务括号内的数字代表权重。 # 权重越高被选中执行的频率就越高。 task(3) # 这个任务的权重是3 def view_items(self): 任务1查看商品列表 # self.client 是HttpUser内置的HttpSession实例用法和requests库非常像 # 它会自动为我们记录请求的响应时间、状态码等数据 with self.client.get(/api/items, catch_responseTrue) as response: # catch_responseTrue 允许我们自定义成功/失败的判断逻辑 if response.status_code 200: # 可以进一步检查响应内容比如判断JSON中某个字段是否存在 if items in response.json(): response.success() else: response.failure(Response did not contain items key) else: response.failure(fStatus code was {response.status_code}) task(1) # 这个任务的权重是1执行频率大约是view_items的1/3 def view_item_detail(self): 任务2查看随机一个商品的详情 这里演示了如何传递路径参数和查询参数。 # 假设商品ID从1到10 item_id random.randint(1, 10) self.client.get(f/api/items/{item_id}, name/api/items/[id]) # 使用name参数对类似的请求进行分组统计。 # 否则Locust会为每个不同的item_id如/items/1, /items/2单独统计图表会非常混乱。 # 用了name之后所有这些请求都会被统计到“/api/items/[id]”这个条目下。 # on_start方法是一个特殊方法每个模拟用户实例在开始执行循环任务之前会先执行一次。 # 常用于模拟用户登录获取认证令牌等前置操作。 def on_start(self): 用户启动时执行比如登录 login_response self.client.post(/api/login, json{username: test, password: 123456}) if login_response.status_code 200: self.token login_response.json().get(token) # 将token添加到后续请求的头部 self.client.headers {Authorization: fBearer {self.token}} else: # 如果登录失败这个用户实例会停止执行任务 print(Login failed for a user) self.stop(forceTrue) # 强制停止这个用户 # 对应的还有on_stop方法在用户停止运行时调用不常用。脚本要点解析HttpUservsUser如果你的压测对象是HTTP/HTTPS服务就继承HttpUser它内置了self.client一个HttpSession对象。如果你要测试其他协议如WebSocket, TCP可以继承基础的User类然后使用其他客户端库如websocket-client但需要自己处理数据的记录。wait_time这是模拟真实性的关键。between(1,3)表示每次任务执行后随机等待1到3秒。还有constant(1)恒定等待1秒和constant_pacing(1)恒定节奏确保任务间隔至少1秒如果任务执行超过1秒则立即开始下一个等选项。task装饰器这是定义用户行为的核心。权重决定了任务的执行概率。上面例子中view_items和view_item_detail的执行概率比为 3:1。self.client这是发起HTTP请求的利器。它支持get,post,put,delete,patch等方法接口设计和requests库高度一致学习成本极低。它自动为每次请求记录性能数据。catch_response与响应验证默认情况下HTTP状态码为2xx或3xx的请求会被记为成功否则记为失败。但有时业务上200的响应也可能意味着失败如返回{“code”: 500, “msg”: “error”}。使用catch_responseTrue上下文管理器可以让你自定义成功/失败的判断逻辑让测试结果更准确。name参数这是新手最容易忽略也最重要的技巧之一。对于带路径参数或查询参数的URL一定要用name参数进行归一化。否则/api/items/1和/api/items/2会被Locust视为两个完全不同的接口导致统计数据分散无法分析该接口的整体性能。使用name后它们被统一归到/api/items/[id]下。on_start用于用户级别的初始化。注意这里的登录操作也会被计入压测请求。如果你的压测场景不关心登录过程或者想排除登录对核心接口的影响可以考虑在脚本外部预先批量生成令牌然后在on_start中直接分配。3.3 运行与观察保存好locustfile.py后打开终端进入脚本所在目录运行locust默认会启动Web UI并监听http://localhost:8089。打开浏览器访问这个地址你会看到Locust的启动界面。Host填写你要压测的目标系统地址例如http://your-api-server.com。Number of users设置你希望模拟的最大总用户数。Spawn rate设置每秒孵化的用户数。例如设置为100表示每秒启动100个用户直到达到总用户数。Start swarming点击开始压测开始后Web UI会自动刷新展示关键图表Statistics所有请求的聚合统计数据表格包括请求类型、路径、请求次数、失败次数、平均响应时间、最小/最大响应时间、以及重要的分位数响应时间如50%中位数95%分位数。Charts实时变化的曲线图包括每秒请求数RPS、响应时间、在线用户数。Failures失败的请求详情包括异常信息是排查问题的第一现场。ExceptionsLocust脚本运行中抛出的Python异常。Download Data可以下载完整的测试报告CSV格式用于后续更细致的分析。实操心得刚开始压测时建议把“Spawn rate”设得低一些比如10。先让系统缓慢增加负载观察各项指标特别是错误率和响应时间的变化曲线是否平滑。如果一开始就全量猛冲可能瞬间把服务打挂你反而得不到有价值的性能曲线比如你不知道系统是从哪个点开始性能急剧下降的。4. 进阶技巧与实战场景配置掌握了基础之后我们来看看如何用Locust应对更复杂的真实场景。4.1 参数化与测试数据管理压测不能总是用同一份数据。比如注册用户用户名必须唯一查询商品商品ID需要多样化。方法一从文件中读取数据这是最常用的方法。假设我们有一个user_credentials.csv文件username,password user1,pass1 user2,pass2 ... ...在Locust脚本中import csv from locust import HttpUser, task, between class ParameterizedUser(HttpUser): wait_time between(1, 2) def on_start(self): # 在类级别读取数据避免每个用户都读一次文件 if not hasattr(ParameterizedUser, user_data): with open(user_credentials.csv, r) as f: reader csv.DictReader(f) ParameterizedUser.user_data list(reader) # 为当前用户实例分配一组数据 self.current_user self.user_data.pop() if self.user_data else None if self.current_user: # 使用分配的数据登录 self.client.post(/login, jsonself.current_user) task def my_task(self): if self.current_user: # 使用对应用户的数据发起请求 self.client.get(f/profile/{self.current_user[username]})方法二使用队列Queue对于需要循环使用或更复杂调度逻辑的数据Python的queue模块很好用。import queue from locust import HttpUser, task, between class QueueUser(HttpUser): wait_time between(1, 2) # 在类层面初始化一个队列 data_queue queue.Queue() classmethod def on_locust_init(cls, environment, **kwargs): 这是一个类方法在所有用户启动前执行一次用于初始化测试数据 for i in range(1000): cls.data_queue.put({item_id: i, search_keyword: fproduct_{i}}) def on_start(self): # 每个用户从队列中取一个数据项如果队列为空则停止 try: self.test_data self.data_queue.get_nowait() except queue.Empty: self.stop(forceTrue) task def search_item(self): if hasattr(self, test_data): self.client.get(f/search?q{self.test_data[search_keyword]}, name/search)注意事项数据队列是进程安全的吗在单机运行模式下由于Locust使用协程所有用户在一个进程内普通的queue.Queue是线程安全的可以用于协程。但在分布式模式下每个Worker进程有自己独立的内存空间队列数据不会共享。此时你需要将测试数据预先加载到每个Worker节点上或者使用外部存储如Redis来共享测试数据状态。4.2 处理关联与状态保持很多接口调用有前后依赖。比如下单需要购物车ID支付需要订单号。class OrderUser(HttpUser): wait_time between(2, 5) host http://shop.example.com task def create_order_flow(self): # 1. 添加商品到购物车 add_to_cart_resp self.client.post(/cart/add, json{product_id: 123, quantity: 1}) if add_to_cart_resp.ok: cart_id add_to_cart_resp.json().get(cart_id) # 2. 创建订单依赖购物车ID create_order_resp self.client.post(/order/create, json{cart_id: cart_id}) if create_order_resp.ok: order_no create_order_resp.json().get(order_no) # 3. 模拟支付依赖订单号 self.client.post(/payment/submit, json{order_no: order_no, amount: 100})关键在于将上一个接口的响应结果提取出来作为下一个接口的请求参数。这完全遵循普通的Python编程逻辑。4.3 分布式压测部署当单台机器无法产生足够压力或者为了避免压测机自身成为瓶颈时就需要分布式压测。步骤准备多台机器确保所有机器Master和Worker都能访问到你的Locust脚本locustfile.py以及任何它依赖的文件如CSV数据文件。可以通过版本控制工具Git同步或使用共享存储如NFS。启动Master节点在其中一台机器上运行以下命令。--master参数表示以Master模式启动。--expect-workers是可选的用于指定期望连接的Worker数量达到后自动开始测试。locust --master --expect-workers 4启动Worker节点在其他的每台机器上运行以下命令。--worker表示以Worker模式启动--master-host指定Master节点的IP地址。locust --worker --master-hostMASTER_IP在Web UI上操作此时访问Master节点的8089端口如http://MASTER_IP:8089你会看到Worker节点连接的状态。之后的操作设置用户数、启动/停止测试和单机模式完全一样都由Master统一控制数据由所有Worker汇总。踩坑实录分布式压测时一个常见的坑是“Address already in use”。这通常是因为Master或Worker进程异常退出后端口没有立即释放。解决方法是更换端口使用--web-port指定另一个端口如--web-port 8090。查找并杀死占用进程lsof -i :8089找到PID然后kill -9 PID。Worker连接不上Master检查防火墙设置确保Worker机器能访问Master机器的8089端口用于通信和5557端口Locust内部消息端口。4.4 非HTTP协议压测Locust的核心是User类HttpUser只是它的一个特化子类。你可以继承User类使用任何Python客户端库来压测其他协议。示例压测一个TCP Socket服务import socket import time from locust import User, task, between, events class SocketUser(User): 自定义User类用于压测TCP Socket服务 wait_time between(0.1, 0.5) # TCP请求通常间隔更短 def on_start(self): 连接Socket服务器 self.client socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: self.client.connect((tcp-server-host, 12345)) self.client.settimeout(5) # 设置超时 except Exception as e: self.environment.runner.quit() # 连接失败停止测试 raise e task def send_message(self): 发送消息并接收响应 message bPING\r\n # 模拟一个简单的协议 start_time time.time() try: self.client.sendall(message) response self.client.recv(1024) if response bPONG\r\n: # 手动记录一个成功的请求 events.request.fire( request_typeTCP, nameping_pong, response_timeint((time.time() - start_time) * 1000), # 毫秒 response_lengthlen(response), exceptionNone, ) else: events.request.fire( request_typeTCP, nameping_pong, response_timeint((time.time() - start_time) * 1000), response_length0, exceptionException(fUnexpected response: {response}), ) except socket.timeout: events.request.fire( request_typeTCP, nameping_pong, response_timeint((time.time() - start_time) * 1000), response_length0, exceptionsocket.timeout(Receive timeout), ) except Exception as e: events.request.fire( request_typeTCP, nameping_pong, response_timeint((time.time() - start_time) * 1000), response_length0, exceptione, ) def on_stop(self): 关闭连接 self.client.close()关键点在于使用locust.events.request.fire()方法手动向Locust报告每次请求的成功与否和响应时间。这样非HTTP的请求也能在Locust的Web UI和统计报告中完美呈现。5. 性能测试策略与结果分析实战工具会用只是第一步更重要的是如何设计测试场景以及如何解读测试结果。5.1 设计科学的压测场景不要一上来就追求“最大并发数”。一个有意义的性能测试通常包含以下几个阶段基准测试Baseline Test在系统空闲或低负载时用很小的并发如1-5个用户运行一段时间获取系统在“最佳状态”下的性能指标响应时间、吞吐量。这个数据将作为后续对比的基准。负载测试Load Test逐步增加并发用户数观察系统性能指标的变化。目标是找到系统在“预期正常负载”下的表现并确认其是否满足性能要求如1000并发下95%的请求响应时间200ms。压力测试Stress Test继续增加负载直到超过系统的预期峰值目的是找出系统的性能瓶颈和极限容量拐点。例如持续增加用户直到错误率超过5%或响应时间变得不可接受此时对应的并发数就是系统的极限。稳定性测试Endurance Test / Soak Test在系统预期负载或略高下长时间如8小时、24小时持续运行。目的是检查系统在长时间运行下是否有内存泄漏、资源逐渐耗尽、性能逐渐下降等问题。尖峰测试Spike Test在短时间内如1分钟内将负载急剧增加到远高于正常水平的数值然后迅速降回正常。模拟秒杀、热点新闻等场景检查系统的弹性恢复能力。在Locust中你可以通过分阶段运行脚本来模拟这些场景或者编写更复杂的逻辑来控制用户行为的变化。5.2 解读Locust数据关键指标看什么压测运行时眼睛要紧盯几个核心指标指标含义关注点RPS (Requests/s)每秒请求数系统吞吐量的直接体现。在负载增加时RPS应平稳上升。如果用户数增加但RPS不升反降或停滞说明系统已到瓶颈。响应时间 (Response Time)请求从发出到收到响应的时间平均响应时间有参考价值但分位数响应时间P95, P99更重要。P95300ms 意味着95%的请求在300ms内完成。这反映了大多数用户的体验。P99则反映了长尾请求的体验。用户数 (Number of Users)当前活跃的并发用户数与你设定的目标是否一致。注意“总用户数”和“孵化率”决定了这个值增长的速度。失败率 (Failures/s)每秒失败的请求数必须密切监控即使响应时间达标但失败率飙升如0.1%也意味着系统已不稳定。需要立刻停止增压分析失败原因超时、5xx错误、业务逻辑失败。异常 (Exceptions)Locust脚本运行异常可能是脚本bug、网络问题、或目标服务返回了意外数据导致解析失败。一个典型的性能拐点分析当你逐步增加用户数时可能会观察到这样的曲线线性增长期用户数增加RPS线性增加响应时间平稳或缓慢上升。系统资源CPU、内存、网络、数据库连接尚有富余。性能拐点当用户数达到某个临界值后RPS增长变缓甚至持平平均响应时间开始明显上升可能是从几十毫秒跳到几百毫秒P95/P99响应时间飙升得更快。此时系统某个资源通常是数据库连接池、线程池、或某个外部依赖已饱和。性能衰减期继续增加用户RPS可能开始下降响应时间急剧恶化失败率尤其是超时错误大幅上升。系统已过载濒临崩溃。你的目标就是通过压测找到第2阶段的那个“拐点”并分析出是哪个组件导致了瓶颈。5.3 结合系统监控定位瓶颈Locust告诉你“系统慢了”或“出错了”但它不能直接告诉你“为什么”。因此压测时必须同时监控被测系统的各项资源指标服务器资源CPU使用率、内存使用率、磁盘I/O读写等待、利用率、网络带宽。应用中间件Web服务器Nginx/Apache的连接数、请求队列长度应用服务器如Tomcat、Gunicorn的线程池/工作进程状态、JVM堆内存对于Java。数据库连接数、慢查询日志、CPU和锁等待情况。缓存Redis/Memcached的内存使用、连接数、命中率。外部依赖调用第三方API的响应时间和成功率。当Locust显示性能下降时立刻去查看这些监控图表。常见瓶颈迹象CPU持续接近100%计算密集型瓶颈可能需要优化代码或扩容。内存使用率不断上升且不释放可能存在内存泄漏。磁盘I/O等待很高数据库或日志写入成为瓶颈。数据库连接池满应用日志中会出现获取连接超时的错误。需要调整连接池大小或优化SQL。网络带宽打满如果是公网传输大量数据如图片、文件可能受带宽限制。6. 常见问题排查与避坑指南这里汇总了一些我在使用Locust过程中踩过的坑和解决方案希望能帮你节省时间。6.1 Locust脚本与运行问题问题1Address already in use错误。原因8089端口被占用。可能是之前的Locust进程没有完全退出。解决换端口启动locust --web-port 8090找到并杀死占用进程lsof -i :8089或netstat -tulpn | grep :8089然后kill -9 PID。Linux/Mac强制使用被占用的端口不推荐locust --web-port 8089 --master --reuse-port问题2压测时RPS很低但CPU占用不高。原因wait_time设置过长用户大部分时间在“思考”没有发请求。检查你的wait_time设置是否合理。响应时间过长如果服务端响应很慢比如每个请求要2秒那么即使用户不停发请求RPS上限也只有用户数 / 平均响应时间。需要先优化服务端。单机性能瓶颈模拟的用户数可能已经达到单台压测机特别是Worker机的网络或端口限制。尝试使用分布式压测。排查在Locust的Web UI上查看“响应时间”是否异常高。同时在压测机上用top,htop或nload等工具监控资源使用情况。问题3ConnectionResetError或大量失败请求。原因服务端主动断开连接可能因为服务端设置了连接超时时间过短或者并发过高导致服务端如Nginx、后端应用的backlog队列满、连接数超限。压测机端口耗尽单个IP的可用端口数有限约6万个。当模拟数万并发且连接保持时间Keep-Alive较长时可能快速耗尽端口。表现为Cannot assign requested address错误。解决调整服务端配置增加Nginx的worker_connections调整后端应用的线程池/连接池大小优化代码减少请求处理时间。在Locust脚本中为HttpUser配置较短的连接超时和较短的Keep-Alive时间或者禁用Keep-Alive但可能增加服务端压力。class MyUser(HttpUser): # 设置连接超时和请求超时 connection_timeout 10.0 network_timeout 10.0 # 或者在请求时单独指定 # self.client.get(/api, timeout10)对于端口耗尽问题使用分布式压测让多台压测机分担连接压力。6.2 测试结果与数据问题问题4统计数据中同一个接口被拆分成很多条目。原因没有使用name参数对动态URL进行归一化。解决如前文所述在所有带路径参数或查询参数的请求中务必使用name参数。# 错误做法会产生 /items/1, /items/2, ... 无数个统计项 self.client.get(f/items/{item_id}) # 正确做法所有请求都统计到 /items/[id] 下 self.client.get(f/items/{item_id}, name/items/[id])问题5如何测试需要鉴权的API方案在on_start中登录如前文示例每个虚拟用户启动时登录一次获取token并存入self.client.headers。这是最常用的方法但登录请求本身也会被压测。预先批量生成Token在压测开始前通过脚本批量调用登录接口生成一批token并写入文件。在Locust脚本的on_start中直接从文件或队列中取一个token来用。这样可以排除登录过程对核心接口压测的影响。使用固定Token如果测试环境允许可以使用一个具有广泛权限的测试账号token。但要注意这不符合真实用户token各不相同的场景可能绕过一些缓存或限流逻辑。问题6如何模拟更复杂的用户思考时间和行为分布方案wait_time不仅可以用between还可以自定义函数。task装饰器也可以放在类上并通过tasks属性以列表形式定义实现更复杂的权重控制。from locust import HttpUser, task, between import random class ComplexUser(HttpUser): # 自定义等待时间70%的概率等待1-3秒30%的概率等待5-10秒模拟深度浏览用户 def custom_wait_time(self): if random.random() 0.7: return random.uniform(1, 3) else: return random.uniform(5, 10) wait_time custom_wait_time # 直接赋值为函数 # 通过tasks属性定义复杂权重 tasks [ task1, # 这是一个函数引用权重默认是1 {task: task2, weight: 3}, # 字典形式指定权重 {task: task3, weight: 2}, ] # 或者使用task装饰器定义在方法上权重更直观 task(5) def high_freq_task(self): ...6.3 分布式与集成问题问题7分布式压测时Worker节点显示为0或连接不稳定。排查网络与防火墙确保所有Worker节点能访问Master节点的8089端口Web UI和5557端口内部通信。命令参数Worker启动命令中的--master-host必须正确指向Master节点的IP或主机名。最好使用IP地址。脚本与依赖一致性所有Worker节点上的Locust版本、Python版本、locustfile.py脚本以及脚本引用的其他文件如CSV数据文件必须完全一致。查看Master日志在Master启动时加上--logfile master.log查看详细日志。问题8如何将Locust集成到CI/CD流水线中方案Locust支持无头模式--headless非常适合自动化。# 基础命令指定用户数、孵化率、运行时间不启动Web UI locust --headless --users 100 --spawn-rate 10 --run-time 5m --hosthttp://your-api.com # 更多实用参数 # --csvresult # 将结果保存为CSV文件result_stats.csv, result_failures.csv等 # --htmlreport.html # 生成HTML格式的报告 # --loglevelINFO/DEBUG # 设置日志级别 # --only-summary # 运行结束后只打印摘要不打印每个请求的统计你可以在Jenkins、GitLab CI、GitHub Actions等工具中在部署完成后自动执行一段这样的Locust命令。然后通过解析退出码非0表示测试失败如因连接错误或分析生成的CSV/HTML报告中的关键指标如失败率是否超过阈值P95响应时间是否达标来决定本次构建是否成功从而实现性能测试的左移和自动化。最后性能测试本身是一个“破坏性”的活动务必在独立的测试环境进行并提前与相关团队运维、DBA、业务方沟通好压测时间窗口和监控预案。清晰的测试目标、严谨的场景设计、实时的监控联动再加上Locust这样得心应手的工具才能让你真正摸清系统的底细为稳定性保驾护航。 Python性能测试实战：从零掌握Locust分布式压测与结果分析
1. 项目概述为什么是Locust如果你是一名开发者或者负责过线上服务的稳定性保障那你一定对“性能测试”这个词不陌生。当你的应用上线前你肯定想知道它能扛住多少用户同时访问响应时间会不会随着用户增多而急剧变慢服务器资源会不会被瞬间打满这些问题靠拍脑袋是没用的必须靠实实在在的压测数据来说话。市面上性能测试工具不少老牌的JMeter功能强大但配置繁琐写个脚本像在画流程图LoadRunner更是重量级选手价格不菲。对于很多以Python技术栈为主的团队或者希望测试脚本能更灵活、更“程序员友好”的开发者来说Locust的出现就像一股清流。Locust是一个用Python写的开源负载测试工具。它的核心思想非常极客你用Python代码来定义用户行为。你想模拟用户登录、浏览商品、下单支付没问题就像写普通的业务逻辑代码一样去写。这种“代码即脚本”的方式让测试逻辑的表达能力直接拉满也让它天然地融入了CI/CD流程。我第一次接触Locust是因为一个紧急的促销活动压测需求当时用JMeter临时拼凑脚本被那些XML配置和元件搞得头大。换成Locust后一个下午就写出了覆盖全链路的压测脚本那种“用代码掌控一切”的感觉瞬间就爱上了。简单来说Locust适合你如果你1熟悉Python2厌倦了图形界面工具的笨重和限制3希望压测脚本能像项目代码一样被版本管理、被复用、被集成4需要快速验证API或Web服务的性能瓶颈。2. Locust的核心架构与工作原理拆解要玩转一个工具先得理解它的大脑和四肢。Locust的架构设计得很清晰理解了它你写脚本和排查问题都会事半功倍。2.1 核心组件蝗虫、蜂巢与王Locust的命名很有意思直译就是“蝗虫”。在它的世界里有三个核心角色User类蝗虫这是你脚本的核心。你需要定义一个继承自HttpUser或更基础的User的类。这个类的每一个实例在压测运行时都代表一个并发用户。你在类里面定义这个用户会做什么比如先访问首页然后登录再查询数据。TaskSet任务集你可以把它理解为用户行为的“场景剧本”。一个User可以包含多个TaskSet用来组织复杂的、有顺序或权重的用户操作。比如你可以定义一个“浏览任务集”只包含查看商品列表、商品详情和一个“购买任务集”包含加购、下单、支付。TaskSet让代码结构更清晰。Master主节点与 Worker从节点这是Locust支持分布式压测的关键。当你需要模拟成千上万的用户时单机可能无法生成足够的压力或者本身成为瓶颈。这时你可以启动一个Master节点和多个Worker节点。Master节点负责协调它不模拟任何用户只负责分发测试任务、收集所有Worker节点的统计数据并托管Web UI那个实时展示图表的管理界面。Worker节点干活的“苦力”。它们接收Master的指令真正地启动和运行你定义的User类实例模拟用户行为并向Master报告自己的数据。这种主从架构让你可以轻松地利用多台机器的资源发起大规模的压力测试。比如用一台配置一般的机器做Master用几台高配的云服务器做Worker就能轻松模拟出数万甚至数十万的并发用户。2.2 工作流程从代码到压力当你运行一个Locust脚本时背后发生了什么脚本加载Locust启动加载你的Python脚本识别出其中定义的User类。用户孵化根据你在Web UI或命令行中设定的用户数--users和孵化速率--spawn-rateLocust会动态地创建User类的实例。每个实例都是一个独立的绿色线程gevent协程而不是操作系统线程因此资源开销极小单机也能模拟数千并发。任务执行每个用户实例协程会独立地、循环地执行你定义的任务tasks。Locust会按照你给任务设置的权重task(3)随机挑选任务执行并在每个任务之间插入一个随机的等待时间通过wait_time属性设置如between(1, 5)表示等待1到5秒以此来模拟真实用户思考、点击的间隔避免产生完全不间断的“机枪”式请求那不符合真实场景。请求发送与统计当执行到具体的HTTP请求调用如self.client.get(/api/data)时Locust会记录下这个请求的发送时间。收到响应后再记录下响应时间、状态码、响应大小等。这些数据会实时汇总。数据聚合与展示所有数据被汇聚到Master单机运行时就是本机进程Locust实时计算并展示在Web UI上包括每秒请求数RPS、响应时间平均、中位数、P95、P99、当前在线用户数等关键指标。注意这里有一个非常重要的细节。Locust的“并发用户数”和你常听到的“每秒请求数RPS”是两个概念。1000个并发用户如果每个用户平均每5秒发一个请求那么RPS大概在200左右。Locust控制的是并发用户的数量和他们的行为模式最终的RPS是结果而不是直接设定的参数。这更符合真实用户场景。3. 从零开始编写你的第一个Locust压测脚本理论说再多不如动手写一行代码。我们来创建一个最经典的场景压测一个简单的用户查询API。3.1 环境准备与安装首先确保你有一个Python环境3.6及以上。使用pip安装Locust非常简单pip install locust安装完成后在命令行输入locust --help如果能显示帮助信息说明安装成功。3.2 脚本结构解剖创建一个名为locustfile.py的文件这是Locust默认寻找的脚本文件名。我们来逐部分解析一个基础脚本from locust import HttpUser, task, between class QuickstartUser(HttpUser): 定义一个模拟用户类继承自HttpUser。 这个类的每个实例在压测中都是一个独立的并发用户。 # 设置用户在每个任务执行后的等待时间范围单位秒 # 这用于模拟真实用户操作之间的间隔避免产生不合理的持续高压力 wait_time between(1, 3) # 使用task装饰器来标记这是一个用户任务括号内的数字代表权重。 # 权重越高被选中执行的频率就越高。 task(3) # 这个任务的权重是3 def view_items(self): 任务1查看商品列表 # self.client 是HttpUser内置的HttpSession实例用法和requests库非常像 # 它会自动为我们记录请求的响应时间、状态码等数据 with self.client.get(/api/items, catch_responseTrue) as response: # catch_responseTrue 允许我们自定义成功/失败的判断逻辑 if response.status_code 200: # 可以进一步检查响应内容比如判断JSON中某个字段是否存在 if items in response.json(): response.success() else: response.failure(Response did not contain items key) else: response.failure(fStatus code was {response.status_code}) task(1) # 这个任务的权重是1执行频率大约是view_items的1/3 def view_item_detail(self): 任务2查看随机一个商品的详情 这里演示了如何传递路径参数和查询参数。 # 假设商品ID从1到10 item_id random.randint(1, 10) self.client.get(f/api/items/{item_id}, name/api/items/[id]) # 使用name参数对类似的请求进行分组统计。 # 否则Locust会为每个不同的item_id如/items/1, /items/2单独统计图表会非常混乱。 # 用了name之后所有这些请求都会被统计到“/api/items/[id]”这个条目下。 # on_start方法是一个特殊方法每个模拟用户实例在开始执行循环任务之前会先执行一次。 # 常用于模拟用户登录获取认证令牌等前置操作。 def on_start(self): 用户启动时执行比如登录 login_response self.client.post(/api/login, json{username: test, password: 123456}) if login_response.status_code 200: self.token login_response.json().get(token) # 将token添加到后续请求的头部 self.client.headers {Authorization: fBearer {self.token}} else: # 如果登录失败这个用户实例会停止执行任务 print(Login failed for a user) self.stop(forceTrue) # 强制停止这个用户 # 对应的还有on_stop方法在用户停止运行时调用不常用。脚本要点解析HttpUservsUser如果你的压测对象是HTTP/HTTPS服务就继承HttpUser它内置了self.client一个HttpSession对象。如果你要测试其他协议如WebSocket, TCP可以继承基础的User类然后使用其他客户端库如websocket-client但需要自己处理数据的记录。wait_time这是模拟真实性的关键。between(1,3)表示每次任务执行后随机等待1到3秒。还有constant(1)恒定等待1秒和constant_pacing(1)恒定节奏确保任务间隔至少1秒如果任务执行超过1秒则立即开始下一个等选项。task装饰器这是定义用户行为的核心。权重决定了任务的执行概率。上面例子中view_items和view_item_detail的执行概率比为 3:1。self.client这是发起HTTP请求的利器。它支持get,post,put,delete,patch等方法接口设计和requests库高度一致学习成本极低。它自动为每次请求记录性能数据。catch_response与响应验证默认情况下HTTP状态码为2xx或3xx的请求会被记为成功否则记为失败。但有时业务上200的响应也可能意味着失败如返回{“code”: 500, “msg”: “error”}。使用catch_responseTrue上下文管理器可以让你自定义成功/失败的判断逻辑让测试结果更准确。name参数这是新手最容易忽略也最重要的技巧之一。对于带路径参数或查询参数的URL一定要用name参数进行归一化。否则/api/items/1和/api/items/2会被Locust视为两个完全不同的接口导致统计数据分散无法分析该接口的整体性能。使用name后它们被统一归到/api/items/[id]下。on_start用于用户级别的初始化。注意这里的登录操作也会被计入压测请求。如果你的压测场景不关心登录过程或者想排除登录对核心接口的影响可以考虑在脚本外部预先批量生成令牌然后在on_start中直接分配。3.3 运行与观察保存好locustfile.py后打开终端进入脚本所在目录运行locust默认会启动Web UI并监听http://localhost:8089。打开浏览器访问这个地址你会看到Locust的启动界面。Host填写你要压测的目标系统地址例如http://your-api-server.com。Number of users设置你希望模拟的最大总用户数。Spawn rate设置每秒孵化的用户数。例如设置为100表示每秒启动100个用户直到达到总用户数。Start swarming点击开始压测开始后Web UI会自动刷新展示关键图表Statistics所有请求的聚合统计数据表格包括请求类型、路径、请求次数、失败次数、平均响应时间、最小/最大响应时间、以及重要的分位数响应时间如50%中位数95%分位数。Charts实时变化的曲线图包括每秒请求数RPS、响应时间、在线用户数。Failures失败的请求详情包括异常信息是排查问题的第一现场。ExceptionsLocust脚本运行中抛出的Python异常。Download Data可以下载完整的测试报告CSV格式用于后续更细致的分析。实操心得刚开始压测时建议把“Spawn rate”设得低一些比如10。先让系统缓慢增加负载观察各项指标特别是错误率和响应时间的变化曲线是否平滑。如果一开始就全量猛冲可能瞬间把服务打挂你反而得不到有价值的性能曲线比如你不知道系统是从哪个点开始性能急剧下降的。4. 进阶技巧与实战场景配置掌握了基础之后我们来看看如何用Locust应对更复杂的真实场景。4.1 参数化与测试数据管理压测不能总是用同一份数据。比如注册用户用户名必须唯一查询商品商品ID需要多样化。方法一从文件中读取数据这是最常用的方法。假设我们有一个user_credentials.csv文件username,password user1,pass1 user2,pass2 ... ...在Locust脚本中import csv from locust import HttpUser, task, between class ParameterizedUser(HttpUser): wait_time between(1, 2) def on_start(self): # 在类级别读取数据避免每个用户都读一次文件 if not hasattr(ParameterizedUser, user_data): with open(user_credentials.csv, r) as f: reader csv.DictReader(f) ParameterizedUser.user_data list(reader) # 为当前用户实例分配一组数据 self.current_user self.user_data.pop() if self.user_data else None if self.current_user: # 使用分配的数据登录 self.client.post(/login, jsonself.current_user) task def my_task(self): if self.current_user: # 使用对应用户的数据发起请求 self.client.get(f/profile/{self.current_user[username]})方法二使用队列Queue对于需要循环使用或更复杂调度逻辑的数据Python的queue模块很好用。import queue from locust import HttpUser, task, between class QueueUser(HttpUser): wait_time between(1, 2) # 在类层面初始化一个队列 data_queue queue.Queue() classmethod def on_locust_init(cls, environment, **kwargs): 这是一个类方法在所有用户启动前执行一次用于初始化测试数据 for i in range(1000): cls.data_queue.put({item_id: i, search_keyword: fproduct_{i}}) def on_start(self): # 每个用户从队列中取一个数据项如果队列为空则停止 try: self.test_data self.data_queue.get_nowait() except queue.Empty: self.stop(forceTrue) task def search_item(self): if hasattr(self, test_data): self.client.get(f/search?q{self.test_data[search_keyword]}, name/search)注意事项数据队列是进程安全的吗在单机运行模式下由于Locust使用协程所有用户在一个进程内普通的queue.Queue是线程安全的可以用于协程。但在分布式模式下每个Worker进程有自己独立的内存空间队列数据不会共享。此时你需要将测试数据预先加载到每个Worker节点上或者使用外部存储如Redis来共享测试数据状态。4.2 处理关联与状态保持很多接口调用有前后依赖。比如下单需要购物车ID支付需要订单号。class OrderUser(HttpUser): wait_time between(2, 5) host http://shop.example.com task def create_order_flow(self): # 1. 添加商品到购物车 add_to_cart_resp self.client.post(/cart/add, json{product_id: 123, quantity: 1}) if add_to_cart_resp.ok: cart_id add_to_cart_resp.json().get(cart_id) # 2. 创建订单依赖购物车ID create_order_resp self.client.post(/order/create, json{cart_id: cart_id}) if create_order_resp.ok: order_no create_order_resp.json().get(order_no) # 3. 模拟支付依赖订单号 self.client.post(/payment/submit, json{order_no: order_no, amount: 100})关键在于将上一个接口的响应结果提取出来作为下一个接口的请求参数。这完全遵循普通的Python编程逻辑。4.3 分布式压测部署当单台机器无法产生足够压力或者为了避免压测机自身成为瓶颈时就需要分布式压测。步骤准备多台机器确保所有机器Master和Worker都能访问到你的Locust脚本locustfile.py以及任何它依赖的文件如CSV数据文件。可以通过版本控制工具Git同步或使用共享存储如NFS。启动Master节点在其中一台机器上运行以下命令。--master参数表示以Master模式启动。--expect-workers是可选的用于指定期望连接的Worker数量达到后自动开始测试。locust --master --expect-workers 4启动Worker节点在其他的每台机器上运行以下命令。--worker表示以Worker模式启动--master-host指定Master节点的IP地址。locust --worker --master-hostMASTER_IP在Web UI上操作此时访问Master节点的8089端口如http://MASTER_IP:8089你会看到Worker节点连接的状态。之后的操作设置用户数、启动/停止测试和单机模式完全一样都由Master统一控制数据由所有Worker汇总。踩坑实录分布式压测时一个常见的坑是“Address already in use”。这通常是因为Master或Worker进程异常退出后端口没有立即释放。解决方法是更换端口使用--web-port指定另一个端口如--web-port 8090。查找并杀死占用进程lsof -i :8089找到PID然后kill -9 PID。Worker连接不上Master检查防火墙设置确保Worker机器能访问Master机器的8089端口用于通信和5557端口Locust内部消息端口。4.4 非HTTP协议压测Locust的核心是User类HttpUser只是它的一个特化子类。你可以继承User类使用任何Python客户端库来压测其他协议。示例压测一个TCP Socket服务import socket import time from locust import User, task, between, events class SocketUser(User): 自定义User类用于压测TCP Socket服务 wait_time between(0.1, 0.5) # TCP请求通常间隔更短 def on_start(self): 连接Socket服务器 self.client socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: self.client.connect((tcp-server-host, 12345)) self.client.settimeout(5) # 设置超时 except Exception as e: self.environment.runner.quit() # 连接失败停止测试 raise e task def send_message(self): 发送消息并接收响应 message bPING\r\n # 模拟一个简单的协议 start_time time.time() try: self.client.sendall(message) response self.client.recv(1024) if response bPONG\r\n: # 手动记录一个成功的请求 events.request.fire( request_typeTCP, nameping_pong, response_timeint((time.time() - start_time) * 1000), # 毫秒 response_lengthlen(response), exceptionNone, ) else: events.request.fire( request_typeTCP, nameping_pong, response_timeint((time.time() - start_time) * 1000), response_length0, exceptionException(fUnexpected response: {response}), ) except socket.timeout: events.request.fire( request_typeTCP, nameping_pong, response_timeint((time.time() - start_time) * 1000), response_length0, exceptionsocket.timeout(Receive timeout), ) except Exception as e: events.request.fire( request_typeTCP, nameping_pong, response_timeint((time.time() - start_time) * 1000), response_length0, exceptione, ) def on_stop(self): 关闭连接 self.client.close()关键点在于使用locust.events.request.fire()方法手动向Locust报告每次请求的成功与否和响应时间。这样非HTTP的请求也能在Locust的Web UI和统计报告中完美呈现。5. 性能测试策略与结果分析实战工具会用只是第一步更重要的是如何设计测试场景以及如何解读测试结果。5.1 设计科学的压测场景不要一上来就追求“最大并发数”。一个有意义的性能测试通常包含以下几个阶段基准测试Baseline Test在系统空闲或低负载时用很小的并发如1-5个用户运行一段时间获取系统在“最佳状态”下的性能指标响应时间、吞吐量。这个数据将作为后续对比的基准。负载测试Load Test逐步增加并发用户数观察系统性能指标的变化。目标是找到系统在“预期正常负载”下的表现并确认其是否满足性能要求如1000并发下95%的请求响应时间200ms。压力测试Stress Test继续增加负载直到超过系统的预期峰值目的是找出系统的性能瓶颈和极限容量拐点。例如持续增加用户直到错误率超过5%或响应时间变得不可接受此时对应的并发数就是系统的极限。稳定性测试Endurance Test / Soak Test在系统预期负载或略高下长时间如8小时、24小时持续运行。目的是检查系统在长时间运行下是否有内存泄漏、资源逐渐耗尽、性能逐渐下降等问题。尖峰测试Spike Test在短时间内如1分钟内将负载急剧增加到远高于正常水平的数值然后迅速降回正常。模拟秒杀、热点新闻等场景检查系统的弹性恢复能力。在Locust中你可以通过分阶段运行脚本来模拟这些场景或者编写更复杂的逻辑来控制用户行为的变化。5.2 解读Locust数据关键指标看什么压测运行时眼睛要紧盯几个核心指标指标含义关注点RPS (Requests/s)每秒请求数系统吞吐量的直接体现。在负载增加时RPS应平稳上升。如果用户数增加但RPS不升反降或停滞说明系统已到瓶颈。响应时间 (Response Time)请求从发出到收到响应的时间平均响应时间有参考价值但分位数响应时间P95, P99更重要。P95300ms 意味着95%的请求在300ms内完成。这反映了大多数用户的体验。P99则反映了长尾请求的体验。用户数 (Number of Users)当前活跃的并发用户数与你设定的目标是否一致。注意“总用户数”和“孵化率”决定了这个值增长的速度。失败率 (Failures/s)每秒失败的请求数必须密切监控即使响应时间达标但失败率飙升如0.1%也意味着系统已不稳定。需要立刻停止增压分析失败原因超时、5xx错误、业务逻辑失败。异常 (Exceptions)Locust脚本运行异常可能是脚本bug、网络问题、或目标服务返回了意外数据导致解析失败。一个典型的性能拐点分析当你逐步增加用户数时可能会观察到这样的曲线线性增长期用户数增加RPS线性增加响应时间平稳或缓慢上升。系统资源CPU、内存、网络、数据库连接尚有富余。性能拐点当用户数达到某个临界值后RPS增长变缓甚至持平平均响应时间开始明显上升可能是从几十毫秒跳到几百毫秒P95/P99响应时间飙升得更快。此时系统某个资源通常是数据库连接池、线程池、或某个外部依赖已饱和。性能衰减期继续增加用户RPS可能开始下降响应时间急剧恶化失败率尤其是超时错误大幅上升。系统已过载濒临崩溃。你的目标就是通过压测找到第2阶段的那个“拐点”并分析出是哪个组件导致了瓶颈。5.3 结合系统监控定位瓶颈Locust告诉你“系统慢了”或“出错了”但它不能直接告诉你“为什么”。因此压测时必须同时监控被测系统的各项资源指标服务器资源CPU使用率、内存使用率、磁盘I/O读写等待、利用率、网络带宽。应用中间件Web服务器Nginx/Apache的连接数、请求队列长度应用服务器如Tomcat、Gunicorn的线程池/工作进程状态、JVM堆内存对于Java。数据库连接数、慢查询日志、CPU和锁等待情况。缓存Redis/Memcached的内存使用、连接数、命中率。外部依赖调用第三方API的响应时间和成功率。当Locust显示性能下降时立刻去查看这些监控图表。常见瓶颈迹象CPU持续接近100%计算密集型瓶颈可能需要优化代码或扩容。内存使用率不断上升且不释放可能存在内存泄漏。磁盘I/O等待很高数据库或日志写入成为瓶颈。数据库连接池满应用日志中会出现获取连接超时的错误。需要调整连接池大小或优化SQL。网络带宽打满如果是公网传输大量数据如图片、文件可能受带宽限制。6. 常见问题排查与避坑指南这里汇总了一些我在使用Locust过程中踩过的坑和解决方案希望能帮你节省时间。6.1 Locust脚本与运行问题问题1Address already in use错误。原因8089端口被占用。可能是之前的Locust进程没有完全退出。解决换端口启动locust --web-port 8090找到并杀死占用进程lsof -i :8089或netstat -tulpn | grep :8089然后kill -9 PID。Linux/Mac强制使用被占用的端口不推荐locust --web-port 8089 --master --reuse-port问题2压测时RPS很低但CPU占用不高。原因wait_time设置过长用户大部分时间在“思考”没有发请求。检查你的wait_time设置是否合理。响应时间过长如果服务端响应很慢比如每个请求要2秒那么即使用户不停发请求RPS上限也只有用户数 / 平均响应时间。需要先优化服务端。单机性能瓶颈模拟的用户数可能已经达到单台压测机特别是Worker机的网络或端口限制。尝试使用分布式压测。排查在Locust的Web UI上查看“响应时间”是否异常高。同时在压测机上用top,htop或nload等工具监控资源使用情况。问题3ConnectionResetError或大量失败请求。原因服务端主动断开连接可能因为服务端设置了连接超时时间过短或者并发过高导致服务端如Nginx、后端应用的backlog队列满、连接数超限。压测机端口耗尽单个IP的可用端口数有限约6万个。当模拟数万并发且连接保持时间Keep-Alive较长时可能快速耗尽端口。表现为Cannot assign requested address错误。解决调整服务端配置增加Nginx的worker_connections调整后端应用的线程池/连接池大小优化代码减少请求处理时间。在Locust脚本中为HttpUser配置较短的连接超时和较短的Keep-Alive时间或者禁用Keep-Alive但可能增加服务端压力。class MyUser(HttpUser): # 设置连接超时和请求超时 connection_timeout 10.0 network_timeout 10.0 # 或者在请求时单独指定 # self.client.get(/api, timeout10)对于端口耗尽问题使用分布式压测让多台压测机分担连接压力。6.2 测试结果与数据问题问题4统计数据中同一个接口被拆分成很多条目。原因没有使用name参数对动态URL进行归一化。解决如前文所述在所有带路径参数或查询参数的请求中务必使用name参数。# 错误做法会产生 /items/1, /items/2, ... 无数个统计项 self.client.get(f/items/{item_id}) # 正确做法所有请求都统计到 /items/[id] 下 self.client.get(f/items/{item_id}, name/items/[id])问题5如何测试需要鉴权的API方案在on_start中登录如前文示例每个虚拟用户启动时登录一次获取token并存入self.client.headers。这是最常用的方法但登录请求本身也会被压测。预先批量生成Token在压测开始前通过脚本批量调用登录接口生成一批token并写入文件。在Locust脚本的on_start中直接从文件或队列中取一个token来用。这样可以排除登录过程对核心接口压测的影响。使用固定Token如果测试环境允许可以使用一个具有广泛权限的测试账号token。但要注意这不符合真实用户token各不相同的场景可能绕过一些缓存或限流逻辑。问题6如何模拟更复杂的用户思考时间和行为分布方案wait_time不仅可以用between还可以自定义函数。task装饰器也可以放在类上并通过tasks属性以列表形式定义实现更复杂的权重控制。from locust import HttpUser, task, between import random class ComplexUser(HttpUser): # 自定义等待时间70%的概率等待1-3秒30%的概率等待5-10秒模拟深度浏览用户 def custom_wait_time(self): if random.random() 0.7: return random.uniform(1, 3) else: return random.uniform(5, 10) wait_time custom_wait_time # 直接赋值为函数 # 通过tasks属性定义复杂权重 tasks [ task1, # 这是一个函数引用权重默认是1 {task: task2, weight: 3}, # 字典形式指定权重 {task: task3, weight: 2}, ] # 或者使用task装饰器定义在方法上权重更直观 task(5) def high_freq_task(self): ...6.3 分布式与集成问题问题7分布式压测时Worker节点显示为0或连接不稳定。排查网络与防火墙确保所有Worker节点能访问Master节点的8089端口Web UI和5557端口内部通信。命令参数Worker启动命令中的--master-host必须正确指向Master节点的IP或主机名。最好使用IP地址。脚本与依赖一致性所有Worker节点上的Locust版本、Python版本、locustfile.py脚本以及脚本引用的其他文件如CSV数据文件必须完全一致。查看Master日志在Master启动时加上--logfile master.log查看详细日志。问题8如何将Locust集成到CI/CD流水线中方案Locust支持无头模式--headless非常适合自动化。# 基础命令指定用户数、孵化率、运行时间不启动Web UI locust --headless --users 100 --spawn-rate 10 --run-time 5m --hosthttp://your-api.com # 更多实用参数 # --csvresult # 将结果保存为CSV文件result_stats.csv, result_failures.csv等 # --htmlreport.html # 生成HTML格式的报告 # --loglevelINFO/DEBUG # 设置日志级别 # --only-summary # 运行结束后只打印摘要不打印每个请求的统计你可以在Jenkins、GitLab CI、GitHub Actions等工具中在部署完成后自动执行一段这样的Locust命令。然后通过解析退出码非0表示测试失败如因连接错误或分析生成的CSV/HTML报告中的关键指标如失败率是否超过阈值P95响应时间是否达标来决定本次构建是否成功从而实现性能测试的左移和自动化。最后性能测试本身是一个“破坏性”的活动务必在独立的测试环境进行并提前与相关团队运维、DBA、业务方沟通好压测时间窗口和监控预案。清晰的测试目标、严谨的场景设计、实时的监控联动再加上Locust这样得心应手的工具才能让你真正摸清系统的底细为稳定性保驾护航。