Python压力测试案例如何模拟高并发请求

wen python案例 2026-06-29 173

本文目录导读：

使用 asyncio + aiohttp（推荐）
使用 threading + requests（简单方案）
使用 Locust 框架（专业方案）
高级方案：混合测试策略
使用建议

我来为您介绍几种Python模拟高并发请求的压力测试方案：

使用 asyncio + aiohttp（推荐）

import asyncio
import aiohttp
import time
from typing import List, Dict
import statistics
class AsyncLoadTester:
    def __init__(self, target_url: str, concurrent_users: int, total_requests: int):
        self.target_url = target_url
        self.concurrent_users = concurrent_users
        self.total_requests = total_requests
        self.results: List[float] = []
    async def single_request(self, session: aiohttp.ClientSession, request_id: int):
        """发送单个请求并记录响应时间"""
        start_time = time.time()
        try:
            async with session.get(self.target_url, timeout=aiohttp.ClientTimeout(total=10)) as response:
                elapsed = time.time() - start_time
                self.results.append(elapsed)
                return request_id, response.status, elapsed
        except Exception as e:
            elapsed = time.time() - start_time
            self.results.append(elapsed)
            return request_id, 0, elapsed
    async def worker(self, session: aiohttp.ClientSession, worker_id: int, requests_per_worker: int):
        """工作线程：处理一定数量的请求"""
        tasks = []
        for i in range(requests_per_worker):
            request_id = f"worker_{worker_id}_req_{i}"
            tasks.append(self.single_request(session, request_id))
        return await asyncio.gather(*tasks)
    async def run_test(self):
        """执行压力测试"""
        print(f"开始压力测试...")
        print(f"目标URL: {self.target_url}")
        print(f"并发数: {self.concurrent_users}")
        print(f"总请求数: {self.total_requests}")
        print("-" * 50)
        start_time = time.time()
        # 计算每个worker处理的请求数
        requests_per_worker = self.total_requests // self.concurrent_users
        async with aiohttp.ClientSession() as session:
            # 创建并发任务
            tasks = []
            for i in range(self.concurrent_users):
                tasks.append(self.worker(session, i, requests_per_worker))
            # 并发执行所有任务
            all_results = await asyncio.gather(*tasks)
        total_time = time.time() - start_time
        # 统计结果
        self.print_statistics(total_time)
    def print_statistics(self, total_time: float):
        """打印测试统计结果"""
        if not self.results:
            print("没有收集到数据")
            return
        sorted_results = sorted(self.results)
        successful_requests = len([r for r in self.results if r > 0])
        failed_requests = len([r for r in self.results if r == 0])
        print("\n=== 压力测试结果 ===")
        print(f"总耗时: {total_time:.2f} 秒")
        print(f"请求总数: {len(self.results)}")
        print(f"成功请求: {successful_requests}")
        print(f"失败请求: {failed_requests}")
        print(f"QPS (每秒请求数): {len(self.results)/total_time:.2f}")
        print(f"平均响应时间: {statistics.mean(self.results)*1000:.2f} ms")
        print(f"中位响应时间: {statistics.median(self.results)*1000:.2f} ms")
        # 百分位响应时间
        for percentile in [90, 95, 99]:
            index = int(len(sorted_results) * percentile / 100)
            print(f"P{percentile} 响应时间: {sorted_results[index]*1000:.2f} ms")
        print(f"最小响应时间: {min(self.results)*1000:.2f} ms")
        print(f"最大响应时间: {max(self.results)*1000:.2f} ms")
# 使用示例
async def main():
    tester = AsyncLoadTester(
        target_url="http://localhost:8000/api/endpoint",
        concurrent_users=50,
        total_requests=1000
    )
    await tester.run_test()
if __name__ == "__main__":
    asyncio.run(main())

使用 threading + requests（简单方案）

import threading
import requests
import time
from queue import Queue
from datetime import datetime
class ThreadedLoadTester:
    def __init__(self, target_url: str, num_threads: int, requests_per_thread: int):
        self.target_url = target_url
        self.num_threads = num_threads
        self.requests_per_thread = requests_per_thread
        self.results_queue = Queue()
        self.success_count = 0
        self.failure_count = 0
        self.lock = threading.Lock()
    def worker(self, thread_id: int):
        """单个工作线程"""
        for i in range(self.requests_per_thread):
            start_time = time.time()
            try:
                response = requests.get(
                    self.target_url, 
                    timeout=5,
                    headers={'User-Agent': 'LoadTester/1.0'}
                )
                elapsed = time.time() - start_time
                with self.lock:
                    if response.status_code == 200:
                        self.success_count += 1
                    else:
                        self.failure_count += 1
                self.results_queue.put({
                    'thread_id': thread_id,
                    'request_id': i,
                    'status': response.status_code,
                    'response_time': elapsed,
                    'timestamp': datetime.now()
                })
            except Exception as e:
                elapsed = time.time() - start_time
                with self.lock:
                    self.failure_count += 1
                self.results_queue.put({
                    'thread_id': thread_id,
                    'request_id': i,
                    'status': 0,
                    'response_time': elapsed,
                    'error': str(e)
                })
    def run_test(self):
        """执行压力测试"""
        print(f"[{datetime.now()}] 开始压力测试")
        print(f"URL: {self.target_url}")
        print(f"线程数: {self.num_threads}")
        print(f"每个线程请求数: {self.requests_per_thread}")
        print(f"总请求数: {self.num_threads * self.requests_per_thread}")
        print("-" * 50)
        threads = []
        start_time = time.time()
        # 创建并启动线程
        for i in range(self.num_threads):
            thread = threading.Thread(target=self.worker, args=(i,))
            threads.append(thread)
            thread.start()
        # 等待所有线程完成
        for thread in threads:
            thread.join()
        total_time = time.time() - start_time
        total_requests = self.success_count + self.failure_count
        print(f"\n[{datetime.now()}] 测试完成")
        print(f"总耗时: {total_time:.2f} 秒")
        print(f"总请求数: {total_requests}")
        print(f"成功请求: {self.success_count}")
        print(f"失败请求: {self.failure_count}")
        print(f"QPS: {total_requests/total_time:.2f}")
        # 计算平均响应时间
        response_times = []
        while not self.results_queue.empty():
            result = self.results_queue.get()
            response_times.append(result['response_time'])
        if response_times:
            avg_response = sum(response_times) / len(response_times)
            print(f"平均响应时间: {avg_response*1000:.2f} ms")
# 使用示例
if __name__ == "__main__":
    tester = ThreadedLoadTester(
        target_url="http://localhost:8000/api/endpoint",
        num_threads=20,
        requests_per_thread=50
    )
    tester.run_test()

使用 Locust 框架（专业方案）

首先安装：pip install locust

创建 locustfile.py：

from locust import HttpUser, task, between, events
import random
class WebsiteUser(HttpUser):
    wait_time = between(0.5, 2.5)  # 等待时间间隔
    def on_start(self):
        """用户开始测试时的初始化"""
        print(f"用户 {self.user_id} 开始测试")
    @task(3)  # 权重3
    def view_homepage(self):
        """查看首页"""
        self.client.get("/")
    @task(2)  # 权重2
    def view_api_endpoint(self):
        """查看API端点"""
        with self.client.get("/api/data", catch_response=True) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"状态码: {response.status_code}")
    @task(1)  # 权重1
    def submit_form(self):
        """提交表单"""
        data = {
            "username": f"user_{random.randint(1000, 9999)}",
            "action": "test"
        }
        self.client.post("/api/submit", json=data)
    @task(1)
    def test_slow_endpoint(self):
        """测试慢接口"""
        with self.client.get("/api/slow", catch_response=True, timeout=30) as response:
            if response.elapsed.total_seconds() > 5:
                response.failure(f"响应时间过长: {response.elapsed.total_seconds():.2f}s")
# 运行命令：locust -f locustfile.py --host=http://localhost:8000
# 然后访问 http://localhost:8089 查看Web界面

高级方案：混合测试策略

import asyncio
import aiohttp
import random
from typing import List, Dict, Any
import json
import time
class AdvancedLoadTester:
    """高级压力测试器，支持混合请求类型"""
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.results = []
    async def send_request(self, session: aiohttp.ClientSession, request_config: Dict):
        """发送配置的请求"""
        method = request_config.get('method', 'GET').upper()
        url = request_config['url']
        headers = request_config.get('headers', {})
        data = request_config.get('data', None)
        start_time = time.time()
        try:
            async with session.request(
                method=method,
                url=url,
                headers=headers,
                json=data if method in ['POST', 'PUT'] else None,
                params=data if method == 'GET' else None,
                timeout=aiohttp.ClientTimeout(total=request_config.get('timeout', 30))
            ) as response:
                elapsed = time.time() - start_time
                result = {
                    'success': response.status < 500,
                    'status_code': response.status,
                    'response_time': elapsed,
                    'url': url,
                    'method': method
                }
                if request_config.get('check_response'):
                    body = await response.text()
                    result['response_size'] = len(body)
                return result
        except Exception as e:
            return {
                'success': False,
                'status_code': 0,
                'response_time': time.time() - start_time,
                'error': str(e),
                'url': url,
                'method': method
            }
    async def simulate_realistic_load(self):
        """模拟真实用户行为"""
        print("开始模拟真实用户负载...")
        async with aiohttp.ClientSession() as session:
            tasks = []
            # 生成混合请求模式
            for i in range(self.config.get('total_users', 100)):
                # 每个用户执行一系列操作
                user_tasks = []
                # 随机选择请求类型
                request_types = [
                    {'method': 'GET', 'url': f"{self.config['base_url']}/api/data"},
                    {'method': 'POST', 'url': f"{self.config['base_url']}/api/submit",
                     'data': {'action': 'create'}},
                    {'method': 'PUT', 'url': f"{self.config['base_url']}/api/update/1"},
                    {'method': 'DELETE', 'url': f"{self.config['base_url']}/api/delete/1"}
                ]
                # 每个用户执行多个操作
                for _ in range(random.randint(1, 5)):
                    chosen_request = random.choice(request_types)
                    user_tasks.append(self.send_request(session, chosen_request))
                    # 模拟用户思考时间
                    think_time = random.uniform(0.5, 2.0)
                    await asyncio.sleep(think_time)
                tasks.extend(user_tasks)
            # 并发执行所有请求
            results = await asyncio.gather(*tasks)
            self.results.extend(results)
    def analyze_results(self):
        """分析测试结果"""
        if not self.results:
            print("没有测试结果")
            return
        successful = [r for r in self.results if r['success']]
        failed = [r for r in self.results if not r['success']]
        response_times = [r['response_time'] for r in self.results]
        response_times.sort()
        print("=== 测试结果分析 ===")
        print(f"总请求数: {len(self.results)}")
        print(f"成功: {len(successful)}")
        print(f"失败: {len(failed)}")
        print(f"成功率: {len(successful)/len(self.results)*100:.2f}%")
        print(f"平均响应时间: {sum(response_times)/len(response_times)*1000:.2f}ms")
        # 百分位分析
        percentiles = [50, 75, 90, 95, 99]
        for p in percentiles:
            index = int(len(response_times) * p / 100)
            print(f"P{p}: {response_times[index]*1000:.2f}ms")
# 使用示例
async def main():
    config = {
        'base_url': 'http://localhost:8000',
        'total_users': 50,
        'ramp_up_time': 10,  # 秒
        'duration': 60,       # 秒
    }
    tester = AdvancedLoadTester(config)
    await tester.simulate_realistic_load()
    tester.analyze_results()
if __name__ == "__main__":
    asyncio.run(main())

使用建议

选择合适的工具：
- 简单测试：使用 threading + requests
- 高性能测试：使用 asyncio + aiohttp
- 专业测试：使用 Locust 框架
- 企业级测试：使用 JMeter + Python脚本
测试注意事项：
- 不要在生产环境直接进行压力测试
- 监控服务器资源使用情况（CPU、内存、网络）
- 从低并发开始逐步增加
- 记录详细的测试日志
最佳实践：
- 模拟真实的用户行为模式
- 包含思考时间和随机延迟
- 测试不同的请求类型和负载模式
- 同时监控客户端和服务端指标

这些方案可以根据您的具体需求进行调整和扩展。

本文地址： https://www.bmjup.com/post/11402.html