本文目录导读：
- 使用 Apache Bench (ab) - 最简单
- 使用 wrk - 性能更好
- Python 脚本 - 最灵活
- 使用 Locust - 专业压测工具
- 使用 k6 - 现代压测工具
- Go语言脚本 - 高性能
- 高级并发控制脚本
- 选择建议
我来介绍几种常用的高并发请求模拟脚本和方法。
使用 Apache Bench (ab) - 最简单
# 发送10000个请求，并发100个 ab -n 10000 -c 100 http://your-api.com/endpoint # 带POST请求 ab -n 1000 -c 50 -p post_data.json -T application/json http://your-api.com/endpoint
使用 wrk - 性能更好
# 安装
brew install wrk # macOS
apt install wrk # Ubuntu
# 使用4个线程，100个连接，持续30秒
wrk -t4 -c100 -d30s http://your-api.com/endpoint
# 带POST请求的Lua脚本
cat > post.lua << EOF
wrk.method = "POST"
wrk.body = '{"key": "value"}'
wrk.headers["Content-Type"] = "application/json"
EOF
wrk -t4 -c100 -d30s -s post.lua http://your-api.com/endpoint
Python 脚本 - 最灵活
使用 aiohttp (异步，高性能)
import asyncio
import aiohttp
import time
async def send_request(session, url):
try:
async with session.get(url) as response:
return await response.text()
except Exception as e:
return f"Error: {e}"
async def concurrent_requests(url, concurrent_count, total_requests):
async with aiohttp.ClientSession() as session:
tasks = []
for _ in range(total_requests):
task = asyncio.ensure_future(send_request(session, url))
tasks.append(task)
# 控制并发数
if len(tasks) >= concurrent_count:
results = await asyncio.gather(*tasks)
tasks = []
# 处理剩余任务
if tasks:
results = await asyncio.gather(*tasks)
return results
# 使用示例
url = "http://localhost:8000/api/test"
concurrent = 100 # 并发数
total = 1000 # 总请求数
start = time.time()
results = asyncio.run(concurrent_requests(url, concurrent, total))
end = time.time()
print(f"完成 {len(results)} 个请求")
print(f"用时: {end - start:.2f} 秒")
print(f"QPS: {len(results) / (end - start):.2f}")
使用 ThreadPoolExecutor (多线程)
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def send_request(url):
try:
response = requests.get(url, timeout=5)
return response.status_code
except Exception as e:
return str(e)
def concurrent_test(url, concurrent_count=100, total_requests=1000):
with ThreadPoolExecutor(max_workers=concurrent_count) as executor:
futures = [executor.submit(send_request, url)
for _ in range(total_requests)]
results = []
for future in as_completed(futures):
results.append(future.result())
return results
# 使用示例
url = "http://localhost:8000/api/test"
start = time.time()
results = concurrent_test(url, 100, 1000)
end = time.time()
success = results.count(200)
failed = len(results) - success
print(f"成功: {success}, 失败: {failed}")
print(f"总用时: {end - start:.2f}秒")
使用 Locust - 专业压测工具
# locustfile.py
from locust import HttpUser, task, between
class WebsiteUser(HttpUser):
wait_time = between(1, 2) # 模拟用户等待时间
@task
def test_endpoint(self):
self.client.get("/api/test")
@task(3) # 权重，执行概率更高
def upload_data(self):
self.client.post("/api/upload",
json={"data": "test"},
headers={"Content-Type": "application/json"})
# 运行
# locust -f locustfile.py --host=http://localhost:8000 --users=100 --spawn-rate=10
使用 k6 - 现代压测工具
// k6-script.js
import http from 'k6/http';
import { sleep, check } from 'k6';
export const options = {
stages: [
{ duration: '30s', target: 100 }, // 逐渐增加到100个并发
{ duration: '1m', target: 100 }, // 保持100并发
{ duration: '30s', target: 0 }, // 逐渐减少
],
thresholds: {
http_req_duration: ['p(95)<500'], // 95%的请求应在500ms以内
},
};
export default function () {
const res = http.get('http://localhost:8000/api/test');
check(res, {
'status is 200': (r) => r.status === 200,
'response time < 200ms': (r) => r.timings.duration < 200,
});
sleep(1);
}
// 运行
// k6 run k6-script.js
Go语言脚本 - 高性能
package main
import (
"fmt"
"net/http"
"sync"
"time"
)
func main() {
url := "http://localhost:8000/api/test"
concurrent := 100
total := 1000
var wg sync.WaitGroup
sem := make(chan struct{}, concurrent) // 控制并发数
start := time.Now()
for i := 0; i < total; i++ {
wg.Add(1)
go func() {
defer wg.Done()
sem <- struct{}{}
defer func() { <-sem }()
resp, err := http.Get(url)
if err != nil {
fmt.Printf("Error: %v\n", err)
return
}
resp.Body.Close()
}()
}
wg.Wait()
elapsed := time.Since(start)
fmt.Printf("完成 %d 个请求\n", total)
fmt.Printf("用时: %v\n", elapsed)
fmt.Printf("QPS: %.2f\n", float64(total)/elapsed.Seconds())
}
高级并发控制脚本
import asyncio
import aiohttp
import time
from collections import Counter
class LoadTester:
def __init__(self, url, concurrent=100, total=1000):
self.url = url
self.concurrent = concurrent
self.total = total
self.results = []
self.errors = []
async def single_request(self, session, semaphore):
async with semaphore:
start = time.time()
try:
async with session.get(self.url, timeout=5) as response:
elapsed = time.time() - start
return {
'status': response.status,
'time': elapsed,
'success': True
}
except Exception as e:
elapsed = time.time() - start
return {
'status': str(e),
'time': elapsed,
'success': False
}
async def run_test(self):
semaphore = asyncio.Semaphore(self.concurrent)
async with aiohttp.ClientSession() as session:
tasks = [self.single_request(session, semaphore)
for _ in range(self.total)]
self.results = await asyncio.gather(*tasks)
def analyze(self):
status_counts = Counter(r['status'] for r in self.results)
times = [r['time'] for r in self.results if r['success']]
print(f"\n=== 测试报告 ===")
print(f"总请求数: {len(self.results)}")
print(f"并发数: {self.concurrent}")
print(f"成功率: {len(times)/len(self.results)*100:.2f}%")
print(f"\n状态码分布:")
for status, count in status_counts.most_common():
print(f" {status}: {count}")
if times:
print(f"\n响应时间:")
print(f" 最小: {min(times):.3f}s")
print(f" 最大: {max(times):.3f}s")
print(f" 平均: {sum(times)/len(times):.3f}s")
times.sort()
print(f" P50: {times[len(times)//2]:.3f}s")
print(f" P90: {times[int(len(times)*0.9)]:.3f}s")
print(f" P99: {times[int(len(times)*0.99)]:.3f}s")
# 使用
tester = LoadTester("http://localhost:8000/api/test",
concurrent=100, total=1000)
asyncio.run(tester.run_test())
tester.analyze()
选择建议
| 工具 | 适用场景 | 性能 | 易用性 |
|---|---|---|---|
| ab | 快速测试 | 一般 | 简单 |
| wrk | 高性能测试 | 高 | 简单 |
| Python aiohttp | 复杂场景 | 中 | 灵活 |
| Locust | 专业压测 | 中 | 图形界面 |
| k6 | CI/CD集成 | 高 | 脚本化 |
| Go | 极致性能 | 最高 | 较复杂 |
最佳实践：
- 先用简单工具快速验证
- 使用Python/Go脚本模拟复杂业务场景
- 最后用Locust/k6进行正式压测
- 注意控制测试负载,避免影响生产环境
