如何直接通过Python代码运行Locust测试（无需终端）

当然可以，你完全可以跳过Shell命令，直接调用Locust的核心API编写内嵌式测试代码，非常适合部署在AWS Lambda这类无终端的环境中做负载测试。以下是具体实现方案：

核心思路 #

Locust本身提供了可直接调用的Python API，通过Environment和Runner类可以手动初始化测试环境、启动测试进程，无需依赖locust命令行工具。

完整代码示例 #

from locust import HttpUser, task, between
from locust.env import Environment
from locust.stats import stats_printer, stats_history
from locust.log import setup_logging
import gevent

# 定义负载测试的用户行为
class APITestUser(HttpUser):
    # 设置请求间隔时间范围
    wait_time = between(1, 3)
    
    @task(1)
    def query_core_api(self):
        # 替换为你的目标API地址
        self.client.get("/api/v1/core-service")
    
    @task(2)
    def submit_data_api(self):
        self.client.post("/api/v1/data-submit", json={"payload": "test-data"})

def run_locust_test():
    # 初始化Locust日志配置
    setup_logging("INFO")
    
    # 创建测试环境，绑定自定义用户类
    env = Environment(user_classes=[APITestUser])
    # 创建本地运行器（Lambda单实例环境用本地模式即可）
    env.create_local_runner()
    
    # 可选：启动统计信息输出（Lambda中会打印到CloudWatch日志）
    gevent.spawn(stats_printer(env.stats))
    gevent.spawn(stats_history, env.runner)
    
    # 启动测试：设置并发用户数和每秒生成速率
    env.runner.start(user_count=20, spawn_rate=5)
    
    # 设置测试运行时长（比如60秒，注意不能超过Lambda的最大超时限制）
    gevent.spawn_later(60, lambda: env.runner.quit())
    
    # 等待测试进程结束
    env.runner.greenlet.join()
    
    # 整理测试结果，可按需扩展字段
    total_stats = env.stats.total
    return {
        "total_requests": total_stats.num_requests,
        "failed_requests": total_stats.num_failures,
        "avg_response_time_ms": round(total_stats.avg_response_time, 2),
        "max_response_time_ms": total_stats.max_response_time,
        "requests_per_second": round(total_stats.total_rps, 2)
    }

# AWS Lambda处理入口函数
def lambda_handler(event, context):
    try:
        test_result = run_locust_test()
        return {
            "statusCode": 200,
            "body": test_result
        }
    except Exception as e:
        return {
            "statusCode": 500,
            "body": f"测试运行失败: {str(e)}"
        }

AWS Lambda部署注意事项 #

• 依赖打包：将Locust及其依赖（如gevent、requests等）打包成Lambda层，或者使用Docker镜像部署Lambda（推荐，避免依赖版本冲突）
• 资源配置：根据并发用户数调整Lambda内存（建议至少512MB，内存越高CPU性能越强），超时时间需覆盖测试运行时长（Lambda最大超时为15分钟）
• 网络权限：确保Lambda能访问到目标API——如果API在VPC内，需将Lambda配置到对应VPC；如果是公网API，需开启Lambda的公网访问权限
• 结果存储：可以将测试结果上传到S3、写入CloudWatch Metrics，方便后续分析和可视化
• 并发限制：注意Lambda的并发配额，避免测试时触发并发限制影响测试结果

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