Locust压测时RPS大幅波动且请求未全部达服务端的原因排查

问题概述 #

• 压测过程中RPS剧烈波动：2秒内从2000骤降至2，与预期行为不符
• 请求丢失：OCI日志显示部分请求未到达服务端，请求处理不一致

附测试脚本：

class UserBehavior(HttpUser):
    host = "https://example.com/api/"

    wait_time = between(2, 2)
    @task
    def one(self):

        actions = [
            (self.client.post, "createg", {"goal": "New Goal", "amount": random.randint(100, 1000)}),
            (self.client.post, "listg"),
        ]
        action = random.choice(actions)
        with self.client.request(action[0].__name__.upper(), action[1], json=action[2] if len(action) > 2 else None, catch_response=True) as response:
            if response.status_code != 200:
                response.failure(f"Failed with status code: {response.status_code}")
    @task
    def two(self):

        actions = [
            (self.client.post, "newd", {"amount": random.randint(100, 500)}),
            (self.client.post, "listd"),
            (self.client.post, "requestd", {"amount": random.randint(50, 200)}),
            (self.client.post, "listw")
        ]
        action = random.choice(actions)
        with self.client.request(action[0].__name__.upper(), action[1], json=action[2] if len(action) > 2 else None, catch_response=True) as response:
            if response.status_code != 200:
                response.failure(f"Failed with status code: {response.status_code}")
    @task
    def three(self):

        actions = [
            (self.client.post, "v1/tickets/create", {"subject": "Issue", "description": "Description"}),
            (self.client.post, "v1/tickets/list"),
            (self.client.post, "v1/tickets/reply", {"message": "Reply"})
        ]
        action = random.choice(actions)
        with self.client.request(action[0].__name__.upper(), action[1], json=action[2] if len(action) > 2 else None, catch_response=True) as response:
            if response.status_code != 200:
                response.failure(f"Failed with status code: {response.status_code}")
env = Environment(user_classes=[UserBehavior])
runner = env.create_local_runner()

web_ui = env.create_web_ui("0.0.0.0", 8090)

env.events.init.fire(environment=env, runner=runner, web_ui=web_ui)

gevent.spawn(stats_history, env.runner)

def start_test():
    num_users = 20000

    spawn_rate = 67  
    runner.start(user_count=num_users, spawn_rate=spawn_rate)
@events.test_start.add_listener
def on_test_start(environment, **kwargs):

    start_test()
    gevent.spawn_later(3600, lambda: runner.quit())
runner.greenlet.join()

web_ui.stop()

原因分析 #

1. 本地Runner资源耗尽：单台机器用本地Runner启动20000个用户，CPU、内存、网络带宽无法支撑大量gevent协程并发调度，导致协程阻塞，RPS骤降。
2. 未设置请求超时：脚本未配置请求超时，服务端响应缓慢或拒绝连接时，请求会无限挂起，占用协程资源，无法发起新请求，最终引发RPS暴跌。
3. 任务逻辑与wait_time不匹配：wait_time固定为2秒，服务端处理能力下降时，大量用户处于等待状态，新请求生成停滞；随机选择action的逻辑可能导致请求分布不均，加剧波动。
4. 服务端瓶颈：服务端过载时会出现连接拒绝、响应超时等情况，部分请求无法成功送达，对应OCI日志中的请求丢失。
5. 启动逻辑问题：on_test_start中直接启动20000用户，spawn_rate=67意味着需约300秒完成用户生成，期间资源消耗持续上升，易触发瓶颈导致RPS波动。

解决建议 #

• 改用分布式压测：放弃本地Runner，采用Locust分布式架构，将20000用户分摊到多台压测机器，避免单台机器资源耗尽。
• 添加请求超时配置：在client.request中加入timeout参数（如timeout=10），防止请求无限挂起，快速释放协程资源。
• 优化wait_time策略：若需稳定RPS，改用constant_pacing替代between，例如constant_pacing(0.001)可控制用户每秒发起指定次数请求（需根据服务端能力调整）；若模拟真实用户行为，可调整between范围避免固定等待时间。
• 调整用户规模与生成速率：先从小规模用户（如1000）测试，逐步增加用户数，找到服务端和压测机器的承载极限；同时提高spawn_rate，避免过长的用户生成周期引发资源波动。
• 优化脚本逻辑：
  • 合并重复的请求处理逻辑，减少冗余，降低出错概率
  • 验证所有POST请求的参数格式是否符合API要求，避免因参数错误被服务端拒绝
  • 补充成功响应标记：对于状态码200的响应，添加response.success()，确保统计数据准确
• 监控关键指标：压测期间监控压测机器的CPU、内存、网络使用率，以及服务端的CPU、内存、连接数、错误率、响应时间等指标，精准定位瓶颈。
• 排查请求丢失环节：开启Locust的DEBUG级日志，查看客户端是否有连接错误、发送失败等记录，结合OCI服务端日志，确认请求丢失是客户端发送失败还是服务端未接收。

内容的提问来源于stack exchange，提问作者RAN