1. 为什么Locust不是“另一个JMeter”——它解决的是并发模型的根本错位很多人第一次听说Locust是在对比JMeter的性能测试选型讨论里。但如果你真把它当成“Python版JMeter”那从第一行代码开始就走偏了。Locust压根没想复刻JMeter那一套基于线程GUIXML配置的老路它要干的是把性能测试这件事从“模拟用户行为”的表层拉回到“模拟真实用户并发本质”的底层。我最早在2019年接手一个电商秒杀接口压测时踩过这个坑用JMeter跑5000并发机器CPU飙到95%结果发现80%的资源耗在了线程上下文切换和HTTP连接池管理上真正发到服务端的请求连3000都不到。后来换Locust重跑同样5000并发单机CPU稳定在40%吞吐量反而高出22%。这不是工具好坏的问题而是底层模型的代差——JMeter用线程模拟用户Locust用协程模拟用户。协程是什么你可以把它理解成“轻量级的、可自主让出CPU的函数”。一个线程里能跑成百上千个协程它们共享线程栈但各自有独立执行上下文。当一个协程发起HTTP请求等待响应时它不阻塞整个线程而是主动交出控制权让调度器立刻切到下一个协程去干活。这就像餐厅里一个服务员线程同时服务20桌客人协程客人点完菜等上菜时服务员转身去帮其他桌下单、倒水、结账而不是站在原地干等。Locust的task装饰器标记的每个方法本质上就是一个这样的“服务员动作”。所以Locust的关键词从来不是“图形界面”或“元件拖拽”而是User类、TaskSet、HttpUser、events.request_success这些概念。它不让你配置“线程组数量”而是让你定义“一个虚拟用户会做什么”然后由框架自动按你设定的spawn_rate每秒启动多少用户和users总用户数去调度协程。这种设计直接绕开了操作系统线程创建/销毁的开销也规避了传统线程模型下连接复用、超时控制、错误恢复的复杂性。提示Locust默认使用gevent作为协程引擎它通过monkey patching劫持Python标准库的socket、ssl、threading等模块把同步IO调用变成异步协程调度。这意味着你写的还是看起来像同步的requests.get()代码但底层早已是事件驱动的非阻塞IO。这也是为什么Locust脚本写起来比asyncio原生代码简单得多——它把协程的复杂性封装在了框架层。你可能会问那为什么不用asyncio因为asyncio要求你全程用await而HTTP客户端库如aiohttp的API风格和requests差异很大学习成本高且生态支持不如requests成熟。Locust选择gevent就是用“无感协程化”降低入门门槛让测试工程师能聚焦在业务逻辑建模上而不是被异步编程范式绊住脚。这也解释了为什么Locust的安装命令是pip install locust而不是pip install asyncio-locust——它刻意保持与Python最主流生态的无缝衔接。你甚至可以直接在Locust脚本里import requests、pandas、json只要不阻塞IO比如别在task里写time.sleep(5)框架就能帮你协程化调度。这种“对开发者友好”的设计哲学正是它能在Python性能测试领域快速崛起的核心原因。2. 从零写出第一个Locust脚本不是写测试而是定义用户行为模型Locust的脚本不是一堆测试用例的集合而是一个用户行为模型的声明式描述。它的核心文件locustfile.py本质上是一份“虚拟用户说明书”。下面我带你手写一个真实的电商搜索接口压测脚本它比官方文档里的hello world更能体现Locust的设计思想。2.1 用户类定义“谁在用你的系统”from locust import HttpUser, task, between import json import random class SearchUser(HttpUser): # 每个用户实例启动后随机等待1到3秒再开始任务 wait_time between(1, 3) def on_start(self): 用户启动时执行一次模拟登录态获取 # 这里可以调用登录接口把token存到self.client中 # self.client.headers.update({Authorization: fBearer {token}}) pass task(3) def search_by_keyword(self): 搜索关键词权重为3比其他任务更频繁 keywords [手机, 笔记本电脑, 蓝牙耳机, 智能手表, 平板电脑] keyword random.choice(keywords) # 构造带参数的URLLocust会自动记录该请求的响应时间、状态码 with self.client.get( f/api/search?keyword{keyword}page1size20, name/api/search [keyword], catch_responseTrue # 允许手动标记成功/失败 ) as response: if response.status_code ! 200: response.failure(fHTTP {response.status_code}) else: try: data response.json() # 验证返回数据结构是否符合预期 if not isinstance(data.get(items), list): response.failure(Response items is not a list) elif len(data[items]) 0: response.failure(Search returned empty result) except json.JSONDecodeError: response.failure(Invalid JSON response) task(1) def search_by_category(self): 按分类搜索权重为1 categories [101, 102, 103, 104] # 分类ID category_id random.choice(categories) with self.client.get( f/api/search/category/{category_id}?page1size20, name/api/search/category [id], catch_responseTrue ) as response: if response.status_code ! 200: response.failure(fHTTP {response.status_code})这段代码里藏着三个关键设计点第一wait_time between(1, 3)不是简单的“停顿”而是用户思考时间Think Time的建模。真实用户不会秒点秒搜他们输入关键词、犹豫、点击搜索这个间隔本身就是负载的一部分。Locust强制你思考这个细节而JMeter需要你在每个Sampler后加一个“定时器”元件容易遗漏。第二task(3)里的数字3是任务权重不是执行次数。它表示search_by_keyword被调度的概率是search_by_category的3倍。这对应现实场景80%的搜索来自关键词输入20%来自分类导航。你不需要写循环或if判断框架自动按比例分发。第三name/api/search [keyword]这个参数至关重要。它把带动态参数的URL聚合成一个逻辑路径。否则Locust会把/api/search?keyword手机和/api/search?keyword笔记本电脑当成两个完全不同的请求导致监控图表里出现上百条散乱的路径根本看不出瓶颈在哪。这是Locust区别于原始HTTP压测工具的“语义聚合”能力。注意catch_responseTrue开启后你才能用response.success()和response.failure()手动控制请求的成功/失败判定。默认情况下只有HTTP状态码2xx才算成功。但很多API用404表示“无结果”用200body.code40001表示业务错误这时就必须手动干预否则压测报告里的成功率会严重失真。2.2 启动与配置命令行即一切Locust没有GUI所有操作都在终端完成。启动命令看似简单实则暗藏玄机# 基础启动监听localhost:8089 locust -f locustfile.py # 指定host被测服务地址避免脚本里硬编码 locust -f locustfile.py --hosthttps://api.example.com # 分布式启动1个master 2个worker # master节点不发请求只收集数据 locust -f locustfile.py --master --web-host0.0.0.0:8089 # worker节点1 locust -f locustfile.py --worker --master-host192.168.1.100 # worker节点2 locust -f locustfile.py --worker --master-host192.168.1.100这里的关键是--host参数。它会自动注入到self.client的base URL里所有self.client.get(/api/xxx)都会拼接成https://api.example.com/api/xxx。这让你的脚本能轻松切换测试环境dev/staging/prod而不用改代码。分布式模式下master节点只做两件事一是提供Web UI图表、控制台二是协调worker的用户数分配。真正的HTTP请求全部由worker进程发出。worker之间完全无状态master宕机不影响压测进行——这正是Locust能轻松突破单机瓶颈的核心架构。3. Locust Web UI背后的实时数据流不只是看数字而是读取系统脉搏Locust的Web界面默认http://localhost:8089常被误认为是个简易监控面板其实它是整套压测系统的实时数据中枢。它展示的每一个数字背后都是一条持续流动的数据管道。3.1 数据采集层从协程到指标的毫秒级映射Locust内部维护着一个全局的RequestStats对象所有self.client.get()调用最终都会经过它。当一个协程发起请求时流程如下self.client.get()被gevent拦截记录起始时间戳T1请求发送协程让出控制权响应到达协程被唤醒记录结束时间戳T2计算耗时Δt T2 - T1并根据状态码、异常类型归类将Δt、status_code、name等字段写入环形缓冲区每秒从缓冲区聚合出请求数、平均响应时间、中位数、95分位、错误率等这个过程在单个worker进程内完成毫秒级延迟。而master节点通过ZeroMQ协议每秒从每个worker拉取一次聚合后的统计快照。这就是为什么你能看到“RPS每秒请求数”曲线如此平滑——它不是采样估算而是精确计数。3.2 关键指标解读超越“响应时间500ms”的粗暴标准新手常盯着“Average Response Time 500ms”这条线但Locust UI里真正值得深挖的是三组数据指标组核心字段诊断价值实操案例吞吐能力RPS、Failures/s、Total Requests系统处理能力的绝对值当RPS不再随用户数线性增长说明已达瓶颈Failures/s突增往往 precede 响应时间飙升响应分布Median、95%、99%、99.9%揭示长尾问题若Average200ms但99%2000ms说明1%的请求严重超时可能是数据库慢查询或缓存击穿错误详情Failures列表、Error Rate定位具体失败类型“ConnectionRefusedError”指向服务端端口耗尽“ReadTimeout”提示下游依赖超时设置过短特别注意“Charts”页签里的响应时间分布直方图。它把响应时间分成多个区间如0-100ms, 100-200ms...显示每个区间内的请求数占比。如果峰值集中在0-100ms但右侧拖着一条长长的尾巴到2000ms这比单纯看平均值更有说服力——说明大部分请求很快但存在偶发性卡顿需要查GC日志或锁竞争。3.3 动态调控在压测中实时调整策略Locust UI右上角的“Start swarming”按钮本质是向master发送一个JSON指令{user_count: 1000, spawn_rate: 10}这触发master按每秒10个的速度向worker分发新用户。你可以随时修改这两个参数增加user_count观察RPS和错误率变化找到拐点调高spawn_rate测试系统瞬时抗压能力如秒杀场景降低user_count验证降级预案是否生效我曾用这个功能发现一个隐蔽问题当用户数从5000降到4000时95%响应时间从800ms骤降至300ms但错误率反而上升了5%。排查发现是服务端熔断阈值设为“错误率3%触发”而降压后流量变均匀原本被高负载掩盖的偶发错误集中暴露出来。这种在压测中动态验证容错机制的能力是静态脚本无法提供的。提示Locust默认每秒刷新一次UI数据但你可以通过--csv参数导出全量原始数据。例如locust -f locustfile.py --csvreport --headless -u 1000 -r 10会生成report_stats.csv里面包含每一秒的详细统计适合用Pandas做深度分析。4. 超越HTTPLocust如何测试WebSocket、gRPC与自定义协议Locust的HttpUser只是开箱即用的特例它的核心是User基类。当你需要测试非HTTP协议时这才是Locust真正展现扩展性的舞台。4.1 WebSocket压测模拟实时聊天场景电商客服系统常用WebSocket维持长连接。用JMeter测试需要装插件而Locust只需几行代码import websocket from locust import User, task, events from locust.exception import RescheduleTask class WebSocketUser(User): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 创建WebSocket连接注意不在__init__里避免worker启动时就连接 self.ws None def on_start(self): 用户启动时建立连接 try: self.ws websocket.WebSocket() self.ws.connect(wss://chat.example.com/ws) # 记录连接建立耗时 events.request_success.fire( request_typewebsocket, nameconnect, response_time0, # WebSocket连接本身不计时后续消息才计 response_length0 ) except Exception as e: events.request_failure.fire( request_typewebsocket, nameconnect, response_time0, exceptione ) task def send_message(self): 发送一条聊天消息 if not self.ws or not self.ws.connected: # 连接断开时主动抛出异常让Locust重试此task raise RescheduleTask() msg {type: message, content: Hello from Locust!} start_time time.time() try: self.ws.send(json.dumps(msg)) # 等待服务器ACK假设协议约定 ack self.ws.recv() total_time int((time.time() - start_time) * 1000) events.request_success.fire( request_typewebsocket, namesend_message, response_timetotal_time, response_lengthlen(ack) ) except Exception as e: total_time int((time.time() - start_time) * 1000) events.request_failure.fire( request_typewebsocket, namesend_message, response_timetotal_time, exceptione ) def on_stop(self): 用户停止时关闭连接 if self.ws: self.ws.close()关键点在于events.request_success/fire()和events.request_failure.fire()这两个手动事件触发。Locust的统计系统不关心你用什么协议只要你按规范上报request_type、name、response_time它就能纳入统一报表。RescheduleTask()异常则让Locust自动重试当前task模拟真实用户在网络抖动时的重连行为。4.2 gRPC压测微服务间调用的精准测量gRPC的二进制协议和流式传输让传统HTTP压测工具束手无策。Locust结合grpcio库可完美应对import grpc from locust import User, task, events import my_pb2 import my_pb2_grpc class GrpcUser(User): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 使用带连接池的channel避免每次task都新建连接 self.channel grpc.insecure_channel(localhost:50051) self.stub my_pb2_grpc.UserServiceStub(self.channel) task def get_user_profile(self): start_time time.time() try: # 发起gRPC调用 request my_pb2.GetUserRequest(user_id123) response self.stub.GetUser(request) total_time int((time.time() - start_time) * 1000) events.request_success.fire( request_typegrpc, nameGetUser, response_timetotal_time, response_lengthlen(response.SerializeToString()) ) except grpc.RpcError as e: total_time int((time.time() - start_time) * 1000) events.request_failure.fire( request_typegrpc, nameGetUser, response_timetotal_time, exceptione ) def on_stop(self): self.channel.close()这里grpc.insecure_channel创建的是一个连接池Locust的每个GrpcUser实例共享同一个channel避免了gRPC连接创建的开销。response.SerializeToString()计算的是序列化后的字节数比HTTP的Content-Length更准确反映网络传输量。4.3 自定义协议TCP/UDP或私有二进制协议对于游戏服务器、IoT设备等使用私有TCP协议的场景Locust同样适用import socket from locust import User, task, events class TcpUser(User): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.sock None def on_start(self): try: self.sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(10) # 设置超时避免阻塞 self.sock.connect((192.168.1.200, 8888)) except Exception as e: events.request_failure.fire( request_typetcp, nameconnect, response_time0, exceptione ) task def send_packet(self): # 构造私有协议包例如4字节长度 JSON body payload b{cmd:heartbeat} packet len(payload).to_bytes(4, big) payload start_time time.time() try: self.sock.sendall(packet) # 接收响应假设协议规定响应长度在前4字节 header self.sock.recv(4) if len(header) 4: raise Exception(Incomplete header) resp_len int.from_bytes(header, big) response self.sock.recv(resp_len) total_time int((time.time() - start_time) * 1000) events.request_success.fire( request_typetcp, namesend_packet, response_timetotal_time, response_lengthlen(response) ) except Exception as e: total_time int((time.time() - start_time) * 1000) events.request_failure.fire( request_typetcp, namesend_packet, response_timetotal_time, exceptione )Locust的扩展性正在于此它不绑定任何协议只提供一个用户生命周期管理框架和统一指标上报接口。你用什么技术实现通信完全由你决定。这种“协议无关”的设计让它能覆盖从Web API到物联网网关的全栈压测场景。5. 生产级实践Locust在CI/CD流水线中的自动化集成Locust的价值不仅在于手动压测更在于成为质量门禁的一部分。把它嵌入CI/CD才能真正实现“每次上线前自动验证性能基线”。5.1 Headless模式脱离UI的自动化执行--headless参数让Locust完全在后台运行适合集成到Jenkins或GitLab CI中# 启动1000用户每秒启动10个运行5分钟生成HTML报告 locust -f locustfile.py \ --headless \ --hosthttps://staging.example.com \ --users 1000 \ --spawn-rate 10 \ --run-time 5m \ --htmlreport.html \ --csvreport关键参数说明--run-time 5m指定压测总时长避免无限运行--htmlreport.html生成交互式HTML报告含图表、表格、下载链接--csvreport生成report_stats.csv、report_failures.csv等原始数据文件生成的report.html可以直接上传到制品库供团队查看。而CSV文件则可用于自动化比对。5.2 性能基线比对用Python脚本自动校验我们写一个简单的校验脚本verify_baseline.py在CI中执行import pandas as pd import sys def check_performance(csv_file, baseline_file): # 读取本次压测结果 df pd.read_csv(csv_file) # 取最后30秒的稳定期数据排除启动阶段波动 stable_df df.tail(30) # 读取基线数据上次发布时保存的report_stats.csv baseline_df pd.read_csv(baseline_file) baseline_last baseline_df.tail(1) # 核心指标比对 current_rps stable_df[Requests/s].mean() baseline_rps baseline_last[Requests/s].iloc[0] current_p95 stable_df[95%].mean() baseline_p95 baseline_last[95%].iloc[0] print(fCurrent RPS: {current_rps:.2f}, Baseline RPS: {baseline_rps:.2f}) print(fCurrent P95: {current_p95:.2f}ms, Baseline P95: {baseline_p95:.2f}ms) # 设定容忍阈值RPS下降不超过5%P95上升不超过10% if current_rps baseline_rps * 0.95: print(❌ RPS下降超过5%性能退化) return False if current_p95 baseline_p95 * 1.10: print(❌ P95上升超过10%响应变慢) return False print(✅ 性能达标) return True if __name__ __main__: if len(sys.argv) ! 3: print(Usage: python verify_baseline.py current_csv baseline_csv) sys.exit(1) success check_performance(sys.argv[1], sys.argv[2]) sys.exit(0 if success else 1)在Jenkins Pipeline中调用stage(Performance Test) { steps { sh locust -f locustfile.py --headless --users 500 --spawn-rate 5 --run-time 3m --csvperf_result sh python verify_baseline.py perf_result_stats.csv baseline_stats.csv } }当校验失败时Pipeline自动中断阻止性能劣化的版本上线。这比人工看报告可靠得多。5.3 分布式集群压测突破单机限制的实战配置单机Locust受限于网络带宽和CPU通常上限在5000-10000并发。要压测百万级用户必须用分布式模式。以下是我在某社交App压测中使用的生产配置硬件准备1台Master4核8G仅作调度不发请求10台Worker每台16核32G千兆网卡所有机器在同一局域网关闭防火墙启动脚本worker.sh#!/bin/bash # 每台worker启动2个Locust进程充分利用多核 locust -f locustfile.py --worker --master-host10.0.1.100 --processes 2 locust -f locustfile.py --worker --master-host10.0.1.100 --processes 2 wait关键优化点--processes 2每个worker启动2个进程每个进程独立使用gevent协程避免单进程GIL限制--expect-workers 10Master启动时预设worker数量避免等待超时--loglevel WARNING降低日志级别减少IO开销压测时Master UI显示“100000 users”10台×2进程×5000用户RPS稳定在8万/s。此时单台worker的CPU使用率约70%网络带宽占用900Mbps证明配置已逼近物理极限。经验Locust分布式模式下worker数量并非越多越好。当worker数超过网络交换机的背板带宽时会出现丢包和延迟激增。我们实测发现10Gbps交换机下20台worker是性价比最优解。超过此数增加worker带来的RPS提升远低于运维复杂度增加。6. 常见陷阱与避坑指南那些Locust文档不会告诉你的事Locust上手容易但要真正用好必须跨过几个隐性门槛。这些坑我都是在给3个不同行业的客户做压测时用真金白银的时间踩出来的。6.1 协程安全陷阱为什么你的脚本在高并发下随机报错最典型的症状是低并发100用户时一切正常升到5000用户后突然出现RuntimeError: dictionary changed size during iteration或AttributeError: NoneType object has no attribute get。根源在于非线程安全的对象被多个协程共享。错误写法# ❌ 全局变量所有协程共用 cache {} task def search(self): key fsearch_{self.keyword} if key not in cache: # 协程A检查时key不存在 cache[key] self._fetch_from_db() # 协程B也在同一时刻执行此行 return cache[key] # 协程A和B都试图读取但cache可能被B修改正确做法用self实例变量每个User实例有独立副本用线程局部存储import threading; local_data threading.local()用协程局部存储推荐from locust.env import Environment; env Environment(); env.cache {}但最根本的解决方案是默认假设所有变量都是协程不安全的除非明确设计为安全。Locust的User类实例是每个虚拟用户独享的所以把状态存在self.xxx里永远安全。6.2 资源泄漏陷阱为什么压测跑着跑着内存就爆了现象压测运行2小时后worker进程内存占用从500MB涨到8GB最终OOM被系统杀死。罪魁祸首往往是未关闭的连接或未释放的资源。典型错误task def download_file(self): # ❌ requests.Session()未关闭连接池累积 session requests.Session() response session.get(https://example.com/large-file.zip) # 忘记session.close()Locust的HttpUser.client已经内置了连接池管理你应该直接用self.client.get()不要自己new Session如果必须用requests务必用with requests.Session() as s:确保关闭对于数据库连接用连接池如SQLAlchemy的create_engine(pool_pre_pingTrue)并设置pool_recycle36006.3 时间精度陷阱为什么你测出的“100ms”实际是200msLocust默认用time.time()获取时间戳但在高并发下Python的time.time()调用本身有微秒级开销且受系统时钟漂移影响。更严重的是gevent的monkey patching会改变time模块行为。实测对比未patch时time.time()精度约15msgevent patch后精度提升至1ms但time.time()在某些Linux内核下仍有抖动解决方案在locustfile.py顶部添加import time # 强制使用更高精度的时钟 if hasattr(time, perf_counter): _time time.perf_counter else: _time time.time # 在task中用_perf_counter() task def api_call(self): start _time() response self.client.get(/api/test) end _time() # 计算耗时...time.perf_counter()是单调递增的高精度计时器不受系统时钟调整影响是测量代码执行时间的黄金标准。6.4 分布式数据同步陷阱为什么Master看不到Worker的自定义指标当你在worker脚本里用events.request_success.fire()上报自定义指标如request_typekafka却发现Master UI里没有显示。这是因为Locust的事件系统默认只同步HTTP相关事件。解决方法在locustfile.py中显式注册自定义事件from locust import events # 在文件顶部注册确保所有worker都加载 events.init.add_listener def on_locust_init(environment, **kwargs): if environment.web_ui: # Master节点注册自定义事件处理器 environment.web_ui.app.view_functions[stats] custom_stats_view更简单的方案是所有自定义指标都用标准的request_type和name字段。Locust的统计系统只认这两个字段只要格式正确如request_typekafkanameproduce_messageMaster就能自动聚合。7. Locust与其他工具的定位对比何时该用它何时该换Locust不是银弹它在性能测试工具光谱中有明确的坐标。理解它的边界比盲目崇拜更重要。7.1 与JMeter的决策树维度LocustJMeter学习成本Python语法即可1小时上手需理解线程组、取样器、监听器等概念2天入门协议支持HTTP/HTTPS原生WebSocket/gRPC/TCP需编码HTTP/FTP/JDBC/JMS原生WebSocket插件gRPC插件分布式扩展原生支持配置简单需配置RMI防火墙复杂易出错报告深度实时图表CSV导出需自行分析内置HTML报告Backend Listener支持InfluxDB/Grafana适用场景团队有Python能力侧重API压测需CI集成测试团队无开发能力需GUI操作压测协议复杂我的建议如果项目组里有至少1个会Python的测试工程师选Locust如果全是功能测试出身且压测需求以HTTP为主JMeter更稳妥。两者并非替代关系而是互补——我们常把Locust用于API层压测JMeter用于端到端流程压测。7.2 与k6的务实选择k6是Go语言编写的现代压测工具性能更强单机10万并发但生态弱于Python。对比关键点性能k6在同等硬件下RPS比Locust高15-20%因Go协程比gevent更轻量生态Locust能直接import requests/pandas/redisk6需用JavaScript生态受限调试Locust用pdb或print调试k6用console.log前者更符合Python工程师习惯选择逻辑追求极致性能且团队熟悉JS选k6需要快速集成现有Python工具链如用pandas分析压测数据选Locust。7.3 与Gatling的错位竞争Gatling是Scala编写的高性能工具强项在DSL脚本和可视化报告。但它要求你学一门新语言Scala且部署依赖JVM。Locust胜在“零学习成本”——你写的压测脚本就是可运行的Python程序能直接复用到生产监控中。举个例子我们曾把Locust脚本稍作修改变成一个常驻进程每5分钟对生产API做健康检查并把结果推送到Prometheus。这种“压测即监控”的能力是Gatling无法提供的。8. 我的Locust实践心得从工具使用者到性能守护者写了这么多技术细节最后想分享一点个人体会。Locust教会我的不是怎么写一个漂亮的压测脚本而是重新理解“性能”这个词的重量。刚接触Locust时我以为性能测试就是跑满CPU、打爆带宽。直到有一次我们用Locust压测一个支付接口RPS、响应时间、错误率全部达标但业务方反馈“用户投诉支付成功率下降”。深入排查才发现问题出在第三方支付网关的连接复用率上——Locust默认的HTTP连接池太激进短时间内建立了大量短连接触发了网关的防刷策略。我们不得不在HttpUser里重写client加入连接复用控制和随机延迟。那一刻我意识到Locust不是魔法棒它只是把系统的真实压力以一种更诚实