脚本怎样模拟高并发请求

wen 实用脚本 2026-06-30 25

本文目录导读：

脚本怎样模拟高并发请求

使用 ab (Apache Bench) 工具
使用 wrk 工具
Python 脚本实现
使用 Locust 框架
Golang 并发脚本
Bash 并发脚本
高级特性脚本
实用建议

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

实用建议

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

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