企业官网建设流程全解析

1. 项目概述：为什么是Locust？

如果你是一名开发者，或者负责过线上服务的稳定性保障，那你一定对“性能测试”这个词不陌生。当你的应用上线前，你肯定想知道：它能扛住多少用户同时访问？响应时间会不会随着用户增多而急剧变慢？服务器资源会不会被瞬间打满？这些问题，靠拍脑袋是没用的，必须靠实实在在的压测数据来说话。

市面上性能测试工具不少，老牌的JMeter功能强大但配置繁琐，写个脚本像在画流程图；LoadRunner更是重量级选手，价格不菲。对于很多以Python技术栈为主的团队，或者希望测试脚本能更灵活、更“程序员友好”的开发者来说，Locust的出现，就像一股清流。

Locust是一个用Python写的开源负载测试工具。它的核心思想非常极客：你用Python代码来定义用户行为。你想模拟用户登录、浏览商品、下单支付？没问题，就像写普通的业务逻辑代码一样去写。这种“代码即脚本”的方式，让测试逻辑的表达能力直接拉满，也让它天然地融入了CI/CD流程。我第一次接触Locust，是因为一个紧急的促销活动压测需求，当时用JMeter临时拼凑脚本，被那些XML配置和元件搞得头大。换成Locust后，一个下午就写出了覆盖全链路的压测脚本，那种“用代码掌控一切”的感觉，瞬间就爱上了。

简单来说，Locust适合你，如果你：1）熟悉Python；2）厌倦了图形界面工具的笨重和限制；3）希望压测脚本能像项目代码一样被版本管理、被复用、被集成；4）需要快速验证API或Web服务的性能瓶颈。

2. Locust的核心架构与工作原理拆解

要玩转一个工具，先得理解它的大脑和四肢。Locust的架构设计得很清晰，理解了它，你写脚本和排查问题都会事半功倍。

2.1 核心组件：蝗虫、蜂巢与王

Locust的命名很有意思，直译就是“蝗虫”。在它的世界里，有三个核心角色：

1. User类（蝗虫）：这是你脚本的核心。你需要定义一个继承自HttpUser（或更基础的User）的类。这个类的每一个实例，在压测运行时都代表一个并发用户。你在类里面定义这个用户会做什么，比如先访问首页，然后登录，再查询数据。
2. TaskSet（任务集）：你可以把它理解为用户行为的“场景剧本”。一个User可以包含多个TaskSet，用来组织复杂的、有顺序或权重的用户操作。比如，你可以定义一个“浏览任务集”（只包含查看商品列表、商品详情）和一个“购买任务集”（包含加购、下单、支付）。TaskSet让代码结构更清晰。
3. Master（主节点）与 Worker（从节点）：这是Locust支持分布式压测的关键。当你需要模拟成千上万的用户时，单机可能无法生成足够的压力，或者本身成为瓶颈。这时，你可以启动一个Master节点和多个Worker节点。
   • Master节点：负责协调，它不模拟任何用户，只负责分发测试任务、收集所有Worker节点的统计数据，并托管Web UI（那个实时展示图表的管理界面）。
   • Worker节点：干活的“苦力”。它们接收Master的指令，真正地启动和运行你定义的User类实例，模拟用户行为，并向Master报告自己的数据。

这种主从架构，让你可以轻松地利用多台机器的资源，发起大规模的压力测试。比如，用一台配置一般的机器做Master，用几台高配的云服务器做Worker，就能轻松模拟出数万甚至数十万的并发用户。

2.2 工作流程：从代码到压力

当你运行一个Locust脚本时，背后发生了什么？

1. 脚本加载：Locust启动，加载你的Python脚本，识别出其中定义的User类。
2. 用户孵化：根据你在Web UI或命令行中设定的用户数（--users）和孵化速率（--spawn-rate），Locust会动态地创建User类的实例。每个实例都是一个独立的绿色线程（gevent协程），而不是操作系统线程，因此资源开销极小，单机也能模拟数千并发。
3. 任务执行：每个用户实例（协程）会独立地、循环地执行你定义的任务（tasks）。Locust会按照你给任务设置的权重（@task(3)）随机挑选任务执行，并在每个任务之间插入一个随机的等待时间（通过wait_time属性设置，如between(1, 5)表示等待1到5秒），以此来模拟真实用户思考、点击的间隔，避免产生完全不间断的“机枪”式请求，那不符合真实场景。
4. 请求发送与统计：当执行到具体的HTTP请求调用（如self.client.get("/api/data")）时，Locust会记录下这个请求的发送时间。收到响应后，再记录下响应时间、状态码、响应大小等。这些数据会实时汇总。
5. 数据聚合与展示：所有数据被汇聚到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_response=True) as response: # catch_response=True 允许我们自定义成功/失败的判断逻辑 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(f"Status 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": f"Bearer {self.token}"} else: # 如果登录失败，这个用户实例会停止执行任务 print("Login failed for a user") self.stop(force=True) # 强制停止这个用户 # 对应的，还有on_stop方法，在用户停止运行时调用（不常用）。

脚本要点解析：

1. HttpUservsUser：如果你的压测对象是HTTP/HTTPS服务，就继承HttpUser，它内置了self.client（一个HttpSession对象）。如果你要测试其他协议（如WebSocket, TCP），可以继承基础的User类，然后使用其他客户端库（如websocket-client），但需要自己处理数据的记录。
2. wait_time：这是模拟真实性的关键。between(1,3)表示每次任务执行后，随机等待1到3秒。还有constant(1)（恒定等待1秒）和constant_pacing(1)（恒定节奏，确保任务间隔至少1秒，如果任务执行超过1秒则立即开始下一个）等选项。
3. @task装饰器：这是定义用户行为的核心。权重决定了任务的执行概率。上面例子中，view_items和view_item_detail的执行概率比为 3:1。
4. self.client：这是发起HTTP请求的利器。它支持get,post,put,delete,patch等方法，接口设计和requests库高度一致，学习成本极低。它自动为每次请求记录性能数据。
5. catch_response与响应验证：默认情况下，HTTP状态码为2xx或3xx的请求会被记为成功，否则记为失败。但有时业务上200的响应也可能意味着失败（如返回{“code”: 500, “msg”: “error”}）。使用catch_response=True上下文管理器，可以让你自定义成功/失败的判断逻辑，让测试结果更准确。
6. name参数：这是新手最容易忽略，也最重要的技巧之一。对于带路径参数或查询参数的URL，一定要用name参数进行归一化。否则，/api/items/1和/api/items/2会被Locust视为两个完全不同的接口，导致统计数据分散，无法分析该接口的整体性能。使用name后，它们被统一归到/api/items/[id]下。
7. on_start：用于用户级别的初始化。注意，这里的登录操作也会被计入压测请求。如果你的压测场景不关心登录过程，或者想排除登录对核心接口的影响，可以考虑在脚本外部预先批量生成令牌，然后在on_start中直接分配。

3.3 运行与观察

保存好locustfile.py后，打开终端，进入脚本所在目录，运行：

locust

默认会启动Web UI，并监听http://localhost:8089。打开浏览器访问这个地址，你会看到Locust的启动界面。

1. Host：填写你要压测的目标系统地址，例如http://your-api-server.com。
2. Number of users：设置你希望模拟的最大总用户数。
3. Spawn rate：设置每秒孵化的用户数。例如，设置为100，表示每秒启动100个用户，直到达到总用户数。
4. Start swarming：点击开始！

压测开始后，Web UI会自动刷新，展示关键图表：

• Statistics：所有请求的聚合统计数据表格，包括请求类型、路径、请求次数、失败次数、平均响应时间、最小/最大响应时间、以及重要的分位数响应时间（如50%中位数，95%分位数）。
• Charts：实时变化的曲线图，包括每秒请求数（RPS）、响应时间、在线用户数。
• Failures：失败的请求详情，包括异常信息，是排查问题的第一现场。
• Exceptions：Locust脚本运行中抛出的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", json=self.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": f"product_{i}"}) def on_start(self): # 每个用户从队列中取一个数据项，如果队列为空则停止 try: self.test_data = self.data_queue.get_nowait() except queue.Empty: self.stop(force=True) @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 分布式压测部署

当单台机器无法产生足够压力，或者为了避免压测机自身成为瓶颈时，就需要分布式压测。

步骤：

1. 准备多台机器：确保所有机器（Master和Worker）都能访问到你的Locust脚本（locustfile.py）以及任何它依赖的文件（如CSV数据文件）。可以通过版本控制工具（Git）同步，或使用共享存储（如NFS）。
2. 启动Master节点：在其中一台机器上，运行以下命令。--master参数表示以Master模式启动。--expect-workers是可选的，用于指定期望连接的Worker数量，达到后自动开始测试。

   locust --master --expect-workers 4

3. 启动Worker节点：在其他的每台机器上，运行以下命令。--worker表示以Worker模式启动，--master-host指定Master节点的IP地址。

   locust --worker --master-host=<MASTER_IP>

4. 在Web UI上操作：此时，访问Master节点的8089端口（如http://<MASTER_IP>:8089），你会看到Worker节点连接的状态。之后的操作（设置用户数、启动/停止测试）和单机模式完全一样，都由Master统一控制，数据由所有Worker汇总。

踩坑实录：分布式压测时，一个常见的坑是“Address already in use”。这通常是因为Master或Worker进程异常退出后，端口没有立即释放。解决方法是：

1. 更换端口：使用--web-port指定另一个端口（如--web-port 8090）。
2. 查找并杀死占用进程：lsof -i :8089找到PID，然后kill -9 PID。
3. 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 = b"PING\r\n" # 模拟一个简单的协议 start_time = time.time() try: self.client.sendall(message) response = self.client.recv(1024) if response == b"PONG\r\n": # 手动记录一个成功的请求 events.request.fire( request_type="TCP", name="ping_pong", response_time=int((time.time() - start_time) * 1000), # 毫秒 response_length=len(response), exception=None, ) else: events.request.fire( request_type="TCP", name="ping_pong", response_time=int((time.time() - start_time) * 1000), response_length=0, exception=Exception(f"Unexpected response: {response}"), ) except socket.timeout: events.request.fire( request_type="TCP", name="ping_pong", response_time=int((time.time() - start_time) * 1000), response_length=0, exception=socket.timeout("Receive timeout"), ) except Exception as e: events.request.fire( request_type="TCP", name="ping_pong", response_time=int((time.time() - start_time) * 1000), response_length=0, exception=e, ) def on_stop(self): """关闭连接""" self.client.close()

关键点在于使用locust.events.request.fire()方法手动向Locust报告每次请求的成功与否和响应时间。这样，非HTTP的请求也能在Locust的Web UI和统计报告中完美呈现。

5. 性能测试策略与结果分析实战

工具会用只是第一步，更重要的是如何设计测试场景，以及如何解读测试结果。

5.1 设计科学的压测场景

不要一上来就追求“最大并发数”。一个有意义的性能测试，通常包含以下几个阶段：

1. 基准测试（Baseline Test）：在系统空闲或低负载时，用很小的并发（如1-5个用户）运行一段时间，获取系统在“最佳状态”下的性能指标（响应时间、吞吐量）。这个数据将作为后续对比的基准。
2. 负载测试（Load Test）：逐步增加并发用户数，观察系统性能指标的变化。目标是找到系统在“预期正常负载”下的表现，并确认其是否满足性能要求（如：1000并发下，95%的请求响应时间<200ms）。
3. 压力测试（Stress Test）：继续增加负载，直到超过系统的预期峰值，目的是找出系统的性能瓶颈和极限容量（拐点）。例如，持续增加用户直到错误率超过5%或响应时间变得不可接受，此时对应的并发数就是系统的极限。
4. 稳定性测试（Endurance Test / Soak Test）：在系统预期负载（或略高）下，长时间（如8小时、24小时）持续运行。目的是检查系统在长时间运行下是否有内存泄漏、资源逐渐耗尽、性能逐渐下降等问题。
5. 尖峰测试（Spike Test）：在短时间内（如1分钟内）将负载急剧增加到远高于正常水平的数值，然后迅速降回正常。模拟秒杀、热点新闻等场景，检查系统的弹性恢复能力。

在Locust中，你可以通过分阶段运行脚本来模拟这些场景，或者编写更复杂的逻辑来控制用户行为的变化。

5.2 解读Locust数据：关键指标看什么？

压测运行时，眼睛要紧盯几个核心指标：

一个典型的性能拐点分析：当你逐步增加用户数时，可能会观察到这样的曲线：

1. 线性增长期：用户数增加，RPS线性增加，响应时间平稳或缓慢上升。系统资源（CPU、内存、网络、数据库连接）尚有富余。
2. 性能拐点：当用户数达到某个临界值后，RPS增长变缓甚至持平，平均响应时间开始明显上升（可能是从几十毫秒跳到几百毫秒），P95/P99响应时间飙升得更快。此时，系统某个资源（通常是数据库连接池、线程池、或某个外部依赖）已饱和。
3. 性能衰减期：继续增加用户，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脚本与运行问题

问题1：Address already in use错误。

• 原因：8089端口被占用。可能是之前的Locust进程没有完全退出。
• 解决：
  1. 换端口启动：locust --web-port 8090
  2. 找到并杀死占用进程：lsof -i :8089或netstat -tulpn | grep :8089，然后kill -9 <PID>。
  3. （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等工具监控资源使用情况。

问题3：ConnectionResetError或大量失败请求。

• 原因：
  1. 服务端主动断开连接：可能因为服务端设置了连接超时时间过短，或者并发过高导致服务端（如Nginx、后端应用）的backlog队列满、连接数超限。
  2. 压测机端口耗尽：单个IP的可用端口数有限（约6万个）。当模拟数万并发且连接保持时间（Keep-Alive）较长时，可能快速耗尽端口。表现为Cannot assign requested address错误。
• 解决：
  1. 调整服务端配置：增加Nginx的worker_connections，调整后端应用的线程池/连接池大小，优化代码减少请求处理时间。
  2. 在Locust脚本中为HttpUser配置较短的连接超时和较短的Keep-Alive时间，或者禁用Keep-Alive（但可能增加服务端压力）。

     class MyUser(HttpUser): # 设置连接超时和请求超时 connection_timeout = 10.0 network_timeout = 10.0 # 或者，在请求时单独指定 # self.client.get("/api", timeout=10)

  3. 对于端口耗尽问题，使用分布式压测，让多台压测机分担连接压力。

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？

• 方案：
  1. 在on_start中登录：如前文示例，每个虚拟用户启动时登录一次，获取token并存入self.client.headers。这是最常用的方法，但登录请求本身也会被压测。
  2. 预先批量生成Token：在压测开始前，通过脚本批量调用登录接口，生成一批token并写入文件。在Locust脚本的on_start中，直接从文件或队列中取一个token来用。这样可以排除登录过程对核心接口压测的影响。
  3. 使用固定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或连接不稳定。

• 排查：
  1. 网络与防火墙：确保所有Worker节点能访问Master节点的8089端口（Web UI）和5557端口（内部通信）。
  2. 命令参数：Worker启动命令中的--master-host必须正确指向Master节点的IP或主机名。最好使用IP地址。
  3. 脚本与依赖一致性：所有Worker节点上的Locust版本、Python版本、locustfile.py脚本以及脚本引用的其他文件（如CSV数据文件）必须完全一致。
  4. 查看Master日志：在Master启动时加上--logfile master.log查看详细日志。

问题8：如何将Locust集成到CI/CD流水线中？

• 方案：Locust支持无头模式（--headless），非常适合自动化。

  # 基础命令：指定用户数、孵化率、运行时间，不启动Web UI locust --headless --users 100 --spawn-rate 10 --run-time 5m --host=http://your-api.com # 更多实用参数： # --csv=result # 将结果保存为CSV文件（result_stats.csv, result_failures.csv等） # --html=report.html # 生成HTML格式的报告 # --loglevel=INFO/DEBUG # 设置日志级别 # --only-summary # 运行结束后只打印摘要，不打印每个请求的统计

  你可以在Jenkins、GitLab CI、GitHub Actions等工具中，在部署完成后自动执行一段这样的Locust命令。然后，通过解析退出码（非0表示测试失败，如因连接错误）或分析生成的CSV/HTML报告中的关键指标（如失败率是否超过阈值，P95响应时间是否达标），来决定本次构建是否成功，从而实现性能测试的左移和自动化。

最后，性能测试本身是一个“破坏性”的活动，务必在独立的测试环境进行，并提前与相关团队（运维、DBA、业务方）沟通好压测时间窗口和监控预案。清晰的测试目标、严谨的场景设计、实时的监控联动，再加上Locust这样得心应手的工具，才能让你真正摸清系统的底细，为稳定性保驾护航。