实用脚本能批量压测吗？

wen 实用脚本 2026-06-09 12

本文目录导读：

实用脚本能批量压测吗？

最简单的 Shell 脚本批量压测
带参数变化的批量压测（Python + aiohttp）
使用 Locust 脚本实现分布式批量压测（工业级）
wrk + Lua 脚本批量压测复杂场景
批量压测 + 自动汇总报表（推荐）
核心建议

可以,用脚本实现批量压测是非常常见的做法，很多压测工具（如 Apache Bench、wrk、Locust）本身就支持脚本化批量执行。

给几个经典实用的批量压测脚本示例（包含 Shell、Python 和 Go）：

最简单的 Shell 脚本批量压测

使用 ab（Apache Bench）或 wrk，配合循环实现不同 URL、不同并发数的批量测试。

示例：20 个不同接口，分别压测 3 次

#!/bin/bash
URLS=(
  "https://api.example.com/login"
  "https://api.example.com/user/1"
  "https://api.example.com/search?q=test"
  # ... 更多 URL
)
for url in "${URLS[@]}"; do
  echo "===== 压测: $url ====="
  for i in 1 2 3; do
    echo "--- 第 $i 次 ---"
    ab -n 1000 -c 50 "$url" 2>&1 | tail -20
    sleep 2  # 每次间隔，防止打挂服务
  done
done

带参数变化的批量压测（Python + aiohttp）

可以动态生成不同请求参数（如用户 ID、随机数据），更适合真实的批量压测。

import asyncio
import aiohttp
import time
async def test_single(session, user_id):
    """压测单个用户的登录/查询接口"""
    url = f"https://api.example.com/user/{user_id}"
    start = time.time()
    try:
        async with session.get(url) as resp:
            await resp.text()
            return time.time() - start
    except:
        return None
async def batch_benchmark(concurrency=50, total_requests=500):
    """批量压测：模拟多个用户并发"""
    connector = aiohttp.TCPConnector(limit=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = []
        for i in range(total_requests):
            user_id = (i % 1000) + 1  # 模拟 1000 个不同用户
            tasks.append(test_single(session, user_id))
        results = await asyncio.gather(*tasks)
        success = [r for r in results if r is not None]
        print(f"总请求: {total_requests}, 成功: {len(success)}, 失败: {len(results) - len(success)}")
        if success:
            print(f"平均耗时: {sum(success)/len(success):.3f}s, 最大: {max(success):.3f}s")
# 执行
asyncio.run(batch_benchmark(concurrency=100, total_requests=1000))

使用 Locust 脚本实现分布式批量压测（工业级）

Locust 本身就是脚本驱动的压测工具，支持批量、分布式。

文件：locustfile.py

from locust import HttpUser, task, between
import random
class BatchTestUser(HttpUser):
    wait_time = between(0.1, 1)  # 模拟用户思考时间
    @task(3)  # 权重3，表示 60% 请求是看首页
    def home(self):
        self.client.get("/")
    @task(2)  # 权重2，40% 是查商品
    def search(self):
        keywords = ["手机", "电脑", "耳机", "键盘"]
        self.client.get(f"/search?q={random.choice(keywords)}")
    @task(1)  # 有时模拟用户登录
    def login(self):
        self.client.post("/login", json={
            "username": f"user_{random.randint(1,1000)}",
            "password": "pass123"
        })

运行命令（批量启动多个 Worker 进程）：

# 主节点（Web 控制台）
locust -f locustfile.py --master --web-port=8089
# 启动 4 个 Worker 进程（可分布在不同机器）
for i in 1 2 3 4; do
  locust -f locustfile.py --worker --master-host=localhost &
done

wrk + Lua 脚本批量压测复杂场景

wrk 支持 Lua 脚本，可以实现动态参数、批量不同 URL。

benchmark.lua

-- 预先准备 100 个不同的 URL
local urls = {}
for i = 1, 100 do
    urls[i] = "/api/item/" .. i
end
request = function()
    local idx = math.random(1, #urls)
    return wrk.format("GET", urls[idx], nil, nil)
end
response = function(status, headers, body)
    -- 可以记录失败
    if status ~= 200 then
        -- 略
    end
end

执行（批量压这 100 个地址）：

wrk -t 20 -c 200 -d 60s -s benchmark.lua https://api.example.com

批量压测 + 自动汇总报表（推荐）

将上述脚本的输出管线化,自动生成 CSV 报告。

#!/bin/bash
echo "url,concurrency,total_requests,avg_time_sec,qps,p99" > report.csv
for url in "https://api1.example.com" "https://api2.example.com"; do
  for concurrency in 10 50 100; do
    echo "压测 $url 并发=$concurrency ..."
    result=$(ab -n 2000 -c $concurrency "$url" 2>&1)
    avg=$(echo "$result" | grep "Time per request" | grep -oP '\d+\.\d+')
    qps=$(echo "$result" | grep "Requests per second" | grep -oP '\d+\.\d+')
    echo "$url,$concurrency,2000,$avg,$qps,-" >> report.csv
  done
done
echo "报告已生成: report.csv"

核心建议

场景	推荐方案	理由
快速批量测几十个静态 URL	Shell + ab/wrk	零依赖，简单
需要参数化、登录态、随机数据	Python + aiohttp	灵活，可控
需要分布式、团队协作、监控图表	Locust	工业级，自带 Web UI
超高并发（10万+ QPS）	wrk + Lua 脚本	性能极佳（C 编写）