脚本怎样模拟高并发请求

本文目录导读：

1. 使用 ab (Apache Bench) 工具
2. 使用 wrk 工具
3. Python 脚本实现
4. 使用 Locust 框架
5. Golang 并发脚本
6. Bash 并发脚本
7. 高级特性脚本
8. 实用建议

使用 ab (Apache Bench) 工具

# 基本用法：-n 请求总数 -c 并发数
ab -n 1000 -c 100 http://example.com/api/test
# 带Headers
ab -n 1000 -c 100 -H "Authorization: Bearer token123" http://example.com/api/test
# POST请求
ab -n 500 -c 50 -p post_data.json -T "application/json" http://example.com/api/submit

使用 wrk 工具

# 基本用法
wrk -t12 -c400 -d30s http://example.com/api/test
# 带脚本
wrk -t4 -c100 -d30s -s script.lua http://example.com/api/test

Python 脚本实现

import requests
import concurrent.futures
import time
import threading
def send_request(url, headers=None):
    """发送单个请求"""
    try:
        start = time.time()
        response = requests.get(url, headers=headers, timeout=5)
        elapsed = time.time() - start
        return {
            'status': response.status_code,
            'time': elapsed,
            'success': True
        }
    except Exception as e:
        return {
            'status': None,
            'time': time.time() - start,
            'success': False,
            'error': str(e)
        }
def concurrent_test(url, concurrent_requests, total_requests):
    """并发测试主函数"""
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=concurrent_requests) as executor:
        futures = []
        for _ in range(total_requests):
            future = executor.submit(send_request, url)
            futures.append(future)
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())
    # 统计结果
    success_count = sum(1 for r in results if r['success'])
    fail_count = total_requests - success_count
    avg_time = sum(r['time'] for r in results) / total_requests
    print(f"总请求数: {total_requests}")
    print(f"成功请求: {success_count}")
    print(f"失败请求: {fail_count}")
    print(f"平均响应时间: {avg_time:.3f}s")
# 使用示例
concurrent_test("http://example.com/api/test", concurrent_requests=50, total_requests=500)

使用 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)  # 权重3
    def test_other_endpoint(self):
        with self.client.post("/api/submit", 
                             json={"key": "value"}, 
                             catch_response=True) as response:
            if response.status_code != 200:
                response.failure("Failed!")

运行：

locust -f locustfile.py --host=http://example.com

Golang 并发脚本

package main
import (
    "fmt"
    "net/http"
    "sync"
    "time"
)
func sendRequest(wg *sync.WaitGroup, url string, results chan<- time.Duration) {
    defer wg.Done()
    start := time.Now()
    resp, err := http.Get(url)
    if err != nil {
        fmt.Printf("Error: %v\n", err)
        return
    }
    defer resp.Body.Close()
    elapsed := time.Since(start)
    results <- elapsed
}
func main() {
    url := "http://example.com/api/test"
    concurrent := 50
    total := 500
    var wg sync.WaitGroup
    results := make(chan time.Duration, total)
    // 控制并发数
    semaphore := make(chan struct{}, concurrent)
    for i := 0; i < total; i++ {
        wg.Add(1)
        semaphore <- struct{}{}
        go func() {
            defer func() { <-semaphore }()
            sendRequest(&wg, url, results)
        }()
    }
    wg.Wait()
    close(results)
    // 统计
    var totalDuration time.Duration
    count := 0
    for duration := range results {
        totalDuration += duration
        count++
    }
    fmt.Printf("总请求数: %d\n", count)
    fmt.Printf("平均响应时间: %v\n", totalDuration/time.Duration(count))
}

Bash 并发脚本

#!/bin/bash
URL="http://example.com/api/test"
CONCURRENT=50
TOTAL=500
concurrent_requests() {
    for i in $(seq 1 $TOTAL); do
        (
            curl -s -o /dev/null -w "%{http_code} %{time_total}\n" $URL
        ) &
        # 控制并发数
        if (( i % CONCURRENT == 0 )); then
            wait
        fi
    done
    wait
}
concurrent_requests

高级特性脚本

import asyncio
import aiohttp
import time
async def async_request(session, url):
    """异步发送请求"""
    try:
        start = time.time()
        async with session.get(url) as response:
            await response.read()
            elapsed = time.time() - start
            return response.status, elapsed
    except Exception as e:
        return None, time.time() - start
async def run_concurrent_test(url, concurrent, total):
    """异步并发测试"""
    timeout = aiohttp.ClientTimeout(total=10)
    connector = aiohttp.TCPConnector(limit=concurrent)
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        tasks = []
        for _ in range(total):
            task = async_request(session, url)
            tasks.append(task)
        results = await asyncio.gather(*tasks)
        success = sum(1 for status, _ in results if status == 200)
        avg_time = sum(elapsed for _, elapsed in results if elapsed) / total
        print(f"Total: {total}, Success: {success}, Fail: {total-success}")
        print(f"Average time: {avg_time:.3f}s")
# 运行
asyncio.run(run_concurrent_test("http://example.com/api/test", 100, 1000))

实用建议

1. 监控服务器资源：同时监控CPU、内存、连接数
2. 逐步增加负载：从低并发逐渐增加
3. 测试不同场景：
   • 正常负载测试
   • 压力测试
   • 突发流量测试
4. 关注指标：
   • 响应时间分布（P50, P95, P99）
   • 错误率
   • 吞吐量（QPS）
   • 资源使用率

选择哪种工具取决于你的具体需求和环境，对于简单测试，ab 或 wrk 就够用；复杂场景建议使用 Locust 或自定义脚本。