如何用脚本模拟高并发请求？

本文目录导读：

1. 使用 Apache Bench (ab) - 最简单
2. 使用 wrk - 性能更好
3. Python 脚本 - 最灵活
4. 使用 Locust - 专业压测工具
5. 使用 k6 - 现代压测工具
6. Go语言脚本 - 高性能
7. 高级并发控制脚本
8. 选择建议

我来介绍几种常用的高并发请求模拟脚本和方法。

使用 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()

选择建议

最佳实践：

1. 先用简单工具快速验证
2. 使用Python/Go脚本模拟复杂业务场景
3. 最后用Locust/k6进行正式压测
4. 注意控制测试负载,避免影响生产环境