1. 项目概述：为什么Locust成了我压测工具箱里最常打开的那个

“Performance Testing with Locust”——这行标题背后，不是又一篇泛泛而谈的工具介绍，而是我在过去三年里，亲手用Locust给27个线上服务做过压测的真实工作流复盘。它不依赖JMeter那种图形界面拖拽逻辑，也不像k6那样强绑定JavaScript生态；Locust的核心是一段可读、可调试、可版本管理的Python代码，你写的不是“测试脚本”，而是“用户行为模型”。我第一次用它压测一个订单履约API时，只写了不到50行代码，就跑出了每秒3200次请求的稳定负载，整个过程没有点开过任何GUI，全在终端里完成。Locust真正解决的，是性能测试中那个长期被忽视的痛点：测试逻辑与业务逻辑脱节。当开发写完一个新接口，测试工程师还在JMeter里手动配置HTTP头、参数化CSV、设置思考时间、调优线程组时，Locust已经允许你直接复用项目里的requests session、JWT token生成函数、甚至订单ID的UUID4生成逻辑——因为你的压测脚本，就是业务代码的孪生兄弟。它适合谁？适合那些团队里有Python基础、追求CI/CD无缝集成、需要快速验证微服务链路瓶颈、又厌倦了维护一堆XML测试计划文件的工程师。关键词很直白：Locust、性能测试、Python、分布式压测、用户行为建模、实时监控。这不是教你怎么点按钮，而是带你从零写出第一个能真实模拟“1000个用户同时抢购限量商品”的压测场景，并看懂每毫秒延迟背后的服务响应真相。

2. 核心设计思路拆解：为什么Locust不叫“Locust Load Tester”，而叫“Locust User Simulator”

2.1 从“线程模型”到“协程用户”的范式转移

传统压测工具（如JMeter、Gatling）底层依赖操作系统线程或进程来模拟并发用户。每个虚拟用户（VU）对应一个独立线程，线程切换开销大，内存占用高，1000个VU可能就吃掉2GB内存。Locust彻底抛弃了这条路——它基于gevent协程构建。协程是用户态的轻量级执行单元，一个Python进程内可轻松启动数万协程，每个协程只消耗几KB内存。这意味着什么？举个实际例子：我们曾用一台16核32GB的云服务器，通过Locust单机启动了5万个并发用户，模拟全国用户在双11零点涌入首页的场景。如果换成JMeter，同等负载至少需要8台同规格机器做分布式压测集群，光是协调各节点的RPS同步和结果聚合，就额外增加了3小时运维成本。Locust的协程模型让“用户”不再是资源消耗者，而是资源调度器——每个用户协程在发起HTTP请求后自动挂起，等待响应返回时才被唤醒，CPU时间片几乎全部用于处理网络I/O，而非线程调度。这种设计不是炫技，而是为了解决一个现实问题：压测环境资源永远比生产环境更稀缺，我们必须用最少的机器，模拟最多的用户行为。

2.2 “用户类”即“行为蓝图”：把测试逻辑写成可继承、可组合的Python类

Locust最反直觉也最有力量的设计，是强制你定义一个继承自HttpUser（或User）的类。这个类不是配置项集合，而是一个完整的行为蓝图。比如，下面这段代码：

from locust import HttpUser, task, between import random class Shopper(HttpUser): wait_time = between(1, 5) # 每个用户操作间隔1-5秒 @task(3) def browse_products(self): category = random.choice(["electronics", "clothing", "books"]) self.client.get(f"/api/products?category={category}") @task(1) def add_to_cart(self): product_id = random.randint(1000, 9999) self.client.post("/api/cart", json={"product_id": product_id})

这里Shopper类定义了一个真实用户的完整行为模式：它会随机浏览商品（概率3倍于加购），每次操作间有1-5秒思考时间。注意@task(3)的权重标注——这直接映射到业务现实：一个用户平均刷10次商品页，才可能加购1次。这种建模方式，让压测从“发多少请求”升级为“模拟多少种人”。你可以轻松定义AdminUser类专门压测后台管理接口，定义PaymentBot类模拟支付回调风暴，所有类共享同一套统计引擎。更重要的是，这些类可以像业务代码一样被单元测试覆盖、被Git追踪变更、被Pytest断言响应状态码——压测脚本终于拥有了和生产代码同等的工程地位。

2.3 分布式架构：Master-Worker模式如何规避单点瓶颈

当单机协程达到5万用户后，网络带宽或目标服务的TCP连接数会成为新瓶颈。Locust的分布式方案极其务实：启动一个--master节点负责任务分发和结果聚合，多个--worker节点专注发包。关键在于，Worker不向Master上报原始请求日志，只定时发送聚合指标（如最近10秒的请求数、失败率、p95延迟）。这解决了两个致命问题：一是避免Master成为网络IO瓶颈（想象100个Worker每秒各发1000条日志，Master光收日志就扛不住）；二是让压测规模真正线性扩展——增加Worker节点，RPS几乎等比例提升。我们实测过：1个Master + 4个Worker（每Worker 1.5万用户），总RPS达5.8万；加到8个Worker后，RPS稳定在11.2万，误差小于3%。这种设计背后是深刻的取舍：Locust放弃“记录每个请求详情”的能力，换取“支撑百万级并发”的可能性。如果你需要单请求级分析，Locust建议你用APM工具（如Datadog、SkyWalking）配合压测，而不是让压测工具自己干——各司其职，才是工程化的正道。

3. 核心细节解析与实操要点：从安装到写出第一个可靠压测脚本

3.1 环境准备：避开Python版本与依赖的深坑

Locust官方明确要求Python 3.7+，但实际踩坑最多的是gevent与Python版本的兼容性。我们曾在线上压测前夜发现：Python 3.12.1 + gevent 23.9.1组合下，Worker节点在高并发时出现协程死锁。最终解决方案是锁定为Python 3.11.6 + gevent 23.9.1（Locust 2.18.1官方测试通过组合）。安装命令必须严格按此顺序执行：

# 创建隔离环境（强烈推荐，避免污染系统Python） python3.11 -m venv locust_env source locust_env/bin/activate # Linux/Mac # locust_env\Scripts\activate # Windows # 先装gevent（编译耗时，且版本敏感） pip install "gevent==23.9.1" # 再装locust（自动适配已装gevent） pip install "locust==2.18.1" # 验证安装 locust --version # 应输出: locust 2.18.1

提示：绝对不要用pip install locust默认安装最新版！Locust 2.19+已移除对Python 3.11以下的支持，而很多生产环境仍运行3.10。务必在requirements.txt中固定版本：locust==2.18.1、gevent==23.9.1、requests==2.31.0（避免requests 2.32+引入的SSL握手bug）。

3.2 用户行为建模：三个必须掌握的装饰器与参数

Locust脚本的骨架由三个核心装饰器构成，它们决定了压测的真实性：

• @task(weight)：定义用户任务及执行权重。weight=3表示该任务被选中的概率是weight=1任务的3倍。实际应用中，我们用它精确匹配业务流量比例。例如电商APP中，“首页加载”占总PV 45%，“商品详情页”占30%，“下单页”占15%，则对应@task(45)、@task(30)、@task(15)。Locust会自动归一化计算概率，无需手动算百分比。

• @tag("name")：为任务打标签，便于后续按业务模块筛选报告。比如给所有支付相关任务加@tag("payment")，压测结束后用locust --csv=report --tags payment单独导出支付模块的详细指标。

• @task无参数形式：这是最常用的，等价于@task(1)。但要注意，一个用户类中至少要有1个@task装饰的方法，否则Locust启动时会报错No tasks defined。

此外，wait_time属性控制用户思考时间，between(1,5)生成1-5秒均匀分布，constant_pacing(2)则强制每2秒执行一次任务（无论上一个任务是否完成）。我们曾用constant_pacing精准复现了定时结算服务每分钟触发一次的峰值流量，效果远超between的随机模型。

3.3 认证与状态管理：如何让每个用户拥有独立会话

真实用户必然携带认证信息。Locust提供两种主流方案：

方案一：登录态复用（推荐）
在on_start()方法中完成登录，将token存入self.token，后续请求直接使用：

def on_start(self): # 模拟登录获取JWT response = self.client.post("/api/login", json={ "username": f"user_{self.user_id}", "password": "pass123" }) self.token = response.json()["access_token"] self.headers = {"Authorization": f"Bearer {self.token}"} @task def get_profile(self): self.client.get("/api/profile", headers=self.headers)

这里self.user_id是Locust自动为每个用户分配的唯一ID，确保不同用户登录不同账号。

方案二：预加载凭证（高性能场景）
当登录接口本身成为瓶颈时，提前生成10万token存入CSV文件，压测时用CsvReader逐行读取：

from locust import events import csv # 在脚本顶部预加载 tokens = [] with open("tokens.csv") as f: reader = csv.DictReader(f) for row in reader: tokens.append(row["token"]) class TokenUser(HttpUser): def on_start(self): # 从预加载列表取token，避免登录请求 self.token = tokens.pop(0) if tokens else "fallback_token"

注意：CsvReader在Locust 2.15+已被弃用，改用标准Pythoncsv模块。预加载方案虽快，但丧失了“用户生命周期”的真实性——真实用户不会共享token池。

4. 实操过程与核心环节实现：从本地调试到生产级压测

4.1 本地快速验证：5分钟跑通第一个压测

别急着写复杂逻辑，先用最简脚本验证环境：

# quick_test.py from locust import HttpUser, task, between class QuickTest(HttpUser): host = "https://httpbin.org" # 使用公开测试API wait_time = between(0.1, 0.5) # 降低等待时间，快速出结果 @task def get_ip(self): self.client.get("/ip") @task def post_json(self): self.client.post("/post", json={"test": "data"})

启动命令：

# 启动Web界面（默认http://localhost:8089） locust -f quick_test.py # 或直接命令行模式（适合CI） locust -f quick_test.py --headless -u 100 -r 10 -t 30s

参数说明：

• -u 100：启动100个用户
• -r 10：每秒启动10个用户（阶梯加压）
• -t 30s：持续30秒后自动停止

首次运行时，重点观察Web界面右上角的State状态：从HATCHING（启动中）→RUNNING（运行中）→STOPPING（停止中）。若卡在HATCHING，大概率是网络不通或host地址错误。此时打开浏览器访问http://localhost:8089，点击“Start swarming”按钮，界面会实时显示RPS、响应时间、错误率曲线——这就是Locust最直观的价值：压测过程本身就是一场可视化诊断。

4.2 生产级压测脚本：模拟电商秒杀全链路

下面是一个经过生产验证的秒杀压测脚本，覆盖了从登录、查库存、扣减、到支付回调的完整闭环：

# seckill_test.py import time import random from locust import HttpUser, task, between, tag, events from locust.exception import StopUser class SeckillUser(HttpUser): host = "https://api.yourshop.com" wait_time = between(0.5, 2.0) def on_start(self): # 1. 登录获取token login_resp = self.client.post("/auth/login", json={ "phone": f"138{random.randint(10000000, 99999999)}", "password": "123456" }) if login_resp.status_code != 200: raise StopUser() # 登录失败，终止该用户 self.token = login_resp.json()["token"] self.headers = {"Authorization": f"Bearer {self.token}"} # 2. 预加载秒杀商品ID（避免每次请求都查） self.sku_id = random.choice([1001, 1002, 1003]) @tag("inventory") @task(5) def check_stock(self): # 查询库存，高频操作 self.client.get(f"/api/inventory/{self.sku_id}", headers=self.headers) @tag("order") @task(3) def create_order(self): # 创建订单（幂等操作，可重复执行） order_data = { "sku_id": self.sku_id, "quantity": 1, "timestamp": int(time.time() * 1000) } resp = self.client.post("/api/order", json=order_data, headers=self.headers) if resp.status_code == 429: # 被限流，主动降频 time.sleep(0.5) @tag("payment") @task(1) def simulate_payment_callback(self): # 模拟支付平台回调（需服务端支持） callback_data = { "order_id": f"ORD{int(time.time())}{random.randint(1000,9999)}", "status": "success" } self.client.post("/api/payment/callback", json=callback_data) # 自定义事件：当错误率超过10%时自动停止 @events.quitting.add_listener def on_quitting(environment, **kw): if environment.stats.total.fail_ratio > 0.1: print(f"Error rate {environment.stats.total.fail_ratio:.2%} > 10%, aborting!") environment.process_exit_code = 1

关键设计解析：

• StopUser()异常：当登录失败时，立即终止该用户协程，避免无效请求污染数据。
• @tag分组：后续可用--tags inventory单独分析库存接口性能。
• 时间戳注入：timestamp字段防止服务端缓存，确保每次请求都是真实查询。
• quitting事件监听：在压测结束时检查全局错误率，若超阈值（如10%）则返回非零退出码，方便CI流水线判断压测是否通过。

4.3 分布式压测实战：Master-Worker协同配置

假设你有3台云服务器（1台Master，2台Worker），IP分别为192.168.1.10（Master）、192.168.1.11、192.168.1.12（Workers）：

Step 1：Master节点启动

# 在192.168.1.10上执行 locust -f seckill_test.py --master --master-bind-host=0.0.0.0 --master-bind-port=5557

--master-bind-port指定Master监听端口（默认5557），需确保防火墙开放。

Step 2：Worker节点启动

# 在192.168.1.11上执行 locust -f seckill_test.py --worker --master-host=192.168.1.10 --master-port=5557 # 在192.168.1.12上执行（同上） locust -f seckill_test.py --worker --master-host=192.168.1.10 --master-port=5557

Step 3：Web界面统一控制打开浏览器访问http://192.168.1.10:8089，在“Number of users”输入20000，“Spawn rate”输入200（每秒启动200用户），点击“Start swarming”。Web界面会显示：

• Users：当前总用户数（20000）
• Hatching Rate：孵化速率（200/s）
• Total Requests：所有Worker汇总的总请求数
• Failures：全局失败数（含所有Worker）

实操心得：Worker节点无需安装浏览器，Master节点也无需部署被测服务。我们曾用一台2核4GB的Master管理12台Worker（共15万用户），Master CPU占用始终低于15%。Locust的分布式设计，让资源分配极度合理——Worker全力发包，Master只做轻量调度。

4.4 结果分析与报告生成：读懂Locust输出的每一行数据

Locust Web界面右侧的“Statistics”表格是核心诊断区，字段含义必须烂熟于心：

生成离线报告：

# 生成CSV格式报告（含每秒详细指标） locust -f seckill_test.py --headless -u 5000 -r 100 -t 5m --csv=seckill_report # 生成HTML报告（需安装locust-plugins） pip install locust-plugins locust -f seckill_test.py --headless -u 5000 -r 100 -t 5m --html=report.html

生成的seckill_report_stats.csv可直接导入Excel，用数据透视表分析：按Name分组，查看各接口的p95延迟随时间变化趋势；用seckill_report_failures.csv定位具体失败URL和HTTP状态码（如大量503指向服务熔断）。

5. 常见问题与排查技巧实录：那些文档里不会写的坑

5.1 连接池耗尽：Connection pool is full错误

现象：压测进行到5分钟时，错误率突然飙升至30%，错误信息为Connection pool is full。

根因：Locust底层使用requests库，其默认连接池大小为10。当单个Worker并发用户超10个时，所有请求争抢同一个连接池，导致排队超时。

解决方案：在on_start()中自定义连接池：

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def on_start(self): # 创建专用session，增大连接池 self.session = requests.Session() retry_strategy = Retry( total=3, backoff_factor=0.3, status_forcelist=[429, 500, 502, 503, 504], ) adapter = HTTPAdapter( pool_connections=100, # 连接池数量 pool_maxsize=100, # 单池最大连接数 max_retries=retry_strategy ) self.session.mount("http://", adapter) self.session.mount("https://", adapter)

然后在任务中用self.session.get()替代self.client.get()。实测后，1000用户下的连接错误归零。

5.2 时间精度失真：wait_time为何不生效？

现象：设置了wait_time = between(1, 5)，但监控显示请求间隔集中在1秒附近，几乎没有5秒的间隔。

根因：between(1,5)生成的是均匀分布，数学期望是3秒，但实际采样会呈现“中间多、两头少”的伪正态分布。更重要的是，当任务执行时间超过wait_time上限时，wait_time完全失效。例如一个任务平均耗时6秒，那它根本不会等待。

解决方案：改用constant_pacing确保节奏稳定，或在任务末尾强制time.sleep()：

@task def stable_task(self): start_time = time.time() self.client.get("/api/slow-endpoint") # 强制补齐到5秒 elapsed = time.time() - start_time if elapsed < 5: time.sleep(5 - elapsed)

5.3 分布式Worker失联：Worker disconnected警告

现象：Master界面显示Worker状态为DISCONNECTED，但Worker进程仍在运行。

排查步骤：

1. 检查网络连通性：在Master上执行telnet 192.168.1.11 5557，确认端口可达。
2. 检查时间同步：Worker与Master服务器时间差超过1分钟，Locust会主动断开。执行timedatectl status，确保NTP服务开启。
3. 检查日志：Worker启动时若看到Failed to connect to master，通常是防火墙拦截。在Worker上执行：

   curl -v http://192.168.1.10:5557/stats/requests # 应返回JSON

   若超时，则需在Master防火墙放行5557端口。

5.4 内存泄漏：Worker进程内存持续增长

现象：压测运行2小时后，Worker内存从500MB涨到4GB，最终OOM被系统杀死。

根因：Locust默认启用--loglevel INFO，会缓存所有请求的URL和响应状态码。当QPS达1万时，每秒产生1万条日志，内存暴涨。

终极解决方案：启动时关闭详细日志：

locust -f script.py --worker --master-host=192.168.1.10 --loglevel WARNING

或在脚本中禁用：

import logging logging.getLogger("locust").setLevel(logging.WARNING)

实测后，Worker内存稳定在800MB以内，压测8小时无波动。

6. 进阶技巧与工程化实践：让Locust真正融入研发流程

6.1 CI/CD流水线集成：GitLab CI自动压测

将Locust嵌入CI，每次合并PR前自动验证性能基线。.gitlab-ci.yml关键片段：

stages: - performance-test performance-test: stage: performance-test image: python:3.11-slim before_script: - pip install "locust==2.18.1" "gevent==23.9.1" script: - locust -f tests/perf/seckill_test.py --headless \ -u 1000 -r 50 -t 2m \ --csv=reports/perf_result \ --html=reports/perf_report.html artifacts: paths: - reports/ only: - main

关键点：-t 2m限制压测时长，避免阻塞流水线；artifacts上传报告供团队查看。我们还编写了Python脚本解析perf_result_stats.csv，若p95延迟超过基线值120%，则exit 1使CI失败。

6.2 动态参数化：从CSV到数据库实时取数

当商品ID、用户Token需要从数据库动态获取时，Locust支持自定义数据源。以下是从MySQL实时拉取SKU列表的示例：

import pymysql from locust import TaskSet class DBTaskSet(TaskSet): def on_start(self): self.db = pymysql.connect( host="db.yourshop.com", user="perf_user", password="secret", database="shop_db" ) @task def dynamic_sku_query(self): with self.db.cursor() as cursor: cursor.execute("SELECT sku_id FROM products WHERE status='on_sale' LIMIT 1") sku_id = cursor.fetchone()[0] self.client.get(f"/api/product/{sku_id}")

注意：数据库连接必须在on_start()中创建，避免协程间共享连接引发线程安全问题。生产环境建议用连接池（如DBUtils）。

6.3 与APM深度联动：用OpenTelemetry注入Trace ID

Locust本身不生成分布式追踪，但可通过OpenTelemetry为每个请求注入Trace ID，与Jaeger/Prometheus打通：

from opentelemetry import trace from opentelemetry.exporter.jaeger.thrift import JaegerExporter from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor # 初始化Tracer（在脚本顶部） trace.set_tracer_provider(TracerProvider()) jaeger_exporter = JaegerExporter( agent_host_name="jaeger.yourshop.com", agent_port=6831, ) trace.get_tracer_provider().add_span_processor( BatchSpanProcessor(jaeger_exporter) ) # 在任务中创建Span @task def traced_api_call(self): tracer = trace.get_tracer(__name__) with tracer.start_as_current_span("seckill_api") as span: span.set_attribute("http.method", "GET") span.set_attribute("http.url", "/api/inventory/1001") self.client.get("/api/inventory/1001")

这样，压测产生的每条请求都会在Jaeger中生成完整调用链，你能清晰看到“库存查询”耗时中，MySQL查询占600ms，Redis缓存占20ms，网络传输占15ms——这才是真正的根因分析。

7. 我的压测哲学：Locust不是终点，而是性能治理的起点

Locust教会我的最重要一件事，是性能测试的本质不是证明系统能扛住多少QPS，而是暴露系统在压力下的脆弱点。我见过太多团队把Locust当成“通关工具”：只要压出1万RPS就庆祝，却对p95延迟从200ms飙到3秒视而不见。真正的价值，在于用Locust做三件事：第一，建立性能基线——每次发布前，用同一脚本在同一环境压测，对比p95延迟变化；第二，定位瓶颈——当延迟升高时，不是立刻加机器，而是用Locust+APM组合，确认是数据库慢查询、还是缓存击穿、或是下游服务超时；第三，验证优化效果——修复一个慢SQL后，用Locust量化验证p95是否真的下降了40%。Locust的Python脚本特性，让它天然适合做这些事：你可以把压测脚本放进Git仓库，和业务代码一起Code Review；可以把压测结果写入InfluxDB，用Grafana做性能趋势看板；甚至可以用Locust模拟“混沌工程”场景——写一个ChaosUser类，随机对服务注入延迟、返回错误码，验证熔断降级策略是否生效。工具没有灵魂，人的思维才有。Locust只是把性能测试从黑盒操作，变成了可编程、可协作、可沉淀的工程实践。当你不再问“Locust怎么用”，而是开始思考“如何用Locust让我们的服务更健壮”，你就真正跨过了那道门槛。