Python使用subprocess执行curl压测报错Argument list too long如何解决

这个错误是你拼接的curl命令长度超过了系统允许的命令行参数最大长度限制（对应系统内核参数ARG_MAX），18732条请求参数拼接后长度动辄几MB到几十MB，远超过默认限制，所以触发报错。此外使用subprocess.run(shell=True)拼接命令的写法本身也存在命令注入风险、解析效率低的问题。

直接用Python的HTTP库发送请求，完全绕过curl命令行长度限制，还能更灵活地统计压测数据，示例如下：

1. 先安装依赖库：pip install requests
2. 压测脚本示例：

import requests
from concurrent.futures import ThreadPoolExecutor

# 读取所有请求参数，此处假设test.txt每行对应一个POST请求的请求体
with open("test.txt", "r", encoding="utf-8") as f:
    payload_list = [line.strip() for line in f if line.strip()]

# 替换为实际接口配置
TARGET_URL = "http://你的目标接口地址"
REQUEST_HEADERS = {
    "Content-Type": "application/json"
}
CONCURRENCY = 30  # 自定义并发数，根据压测需求调整

def send_request(payload):
    try:
        response = requests.post(
            url=TARGET_URL,
            data=payload,
            headers=REQUEST_HEADERS,
            timeout=15
        )
        # 可自行添加响应校验、结果统计逻辑
        return response.status_code
    except Exception as e:
        return f"ERROR: {str(e)}"

if __name__ == "__main__":
    # 多并发执行请求
    with ThreadPoolExecutor(max_workers=CONCURRENCY) as executor:
        result_list = list(executor.map(send_request, payload_list))
    
    # 可自行添加压测结果统计逻辑，如成功率、平均响应时间等
    success_count = sum(1 for res in result_list if isinstance(res, int) and 200 <= res < 400)
    print(f"压测完成：总请求数{len(result_list)}，成功数{success_count}，成功率{success_count/len(result_list)*100:.2f}%")

如果必须保留curl调用的逻辑，按以下规则修改即可避免参数过长问题：

• 关闭shell=True，将curl参数以列表形式传入subprocess.run
• 不要把请求体拼接在命令行中，通过标准输入传递给curl，示例如下：

import subprocess
from concurrent.futures import ThreadPoolExecutor

with open("test.txt", "r", encoding="utf-8") as f:
    payload_list = [line.strip() for line in f if line.strip()]

CONCURRENCY = 30
TARGET_URL = "http://你的目标接口地址"

def run_curl(payload):
    try:
        # @- 表示curl从标准输入读取请求体内容
        proc = subprocess.run(
            [
                "curl", "-X", "POST",
                "-H", "Content-Type: application/json",
                "-d", "@-",
                TARGET_URL
            ],
            input=payload.encode("utf-8"),
            capture_output=True,
            timeout=15
        )
        return proc.returncode
    except Exception as e:
        return f"ERROR: {str(e)}"

if __name__ == "__main__":
    with ThreadPoolExecutor(max_workers=CONCURRENCY) as executor:
        result_list = list(executor.map(run_curl, payload_list))

• 压测量级较大的场景，更推荐使用专业压测工具（如wrk2、JMeter），性能和统计能力都优于自定义脚本
• 无论用哪种方式，都不要将用户可控的参数直接拼接进shell命令，避免命令注入漏洞

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