性能测试（PerformanceTest）使用方法与实践指南

// 性能指标数据模型示例
class PerformanceMetrics {
    constructor() {
        this.responseTime = {
            min: 0,           // 最小响应时间
            max: 0,           // 最大响应时间
            avg: 0,           // 平均响应时间
            p50: 0,           // 中位数响应时间
            p90: 0,           // 90%分位数响应时间
            p95: 0,           // 95%分位数响应时间
            p99: 0            // 99%分位数响应时间
        };
        this.throughput = {
            rps: 0,           // 每秒请求数
            tps: 0,           // 每秒事务数
            totalRequests: 0  // 总请求数
        };
        this.errorRate = 0;   // 错误率
        this.concurrentUsers = 0; // 并发用户数
    }
}

<!-- TRAE IDE 会自动提示 JMeter 元素结构和属性 -->
<ThreadGroup guiclass="ThreadGroupGui" testclass="ThreadGroup" testname="性能测试线程组">
    <stringProp name="ThreadGroup.num_threads">${__P(threads,100)}</stringProp>
    <stringProp name="ThreadGroup.ramp_time">${__P(rampup,300)}</stringProp>
    <stringProp name="ThreadGroup.duration">${__P(duration,1800)}</stringProp>
</ThreadGroup>

# api_performance_test.py
import asyncio
import aiohttp
import time
import statistics
from concurrent.futures import ThreadPoolExecutor
import matplotlib.pyplot as plt
 
class APIPerformanceTester:
    def __init__(self, base_url, concurrent_users=100, total_requests=1000):
        self.base_url = base_url
        self.concurrent_users = concurrent_users
        self.total_requests = total_requests
        self.response_times = []
        self.error_count = 0
        
    async def single_request(self, session, user_id):
        """执行单个API请求"""
        start_time = time.time()
        try:
            async with session.get(f"{self.base_url}/api/users/{user_id}") as response:
                await response.json()
                response_time = (time.time() - start_time) * 1000  # 转换为毫秒
                self.response_times.append(response_time)
                return response.status == 200
        except Exception as e:
            self.error_count += 1
            print(f"请求失败: {e}")
            return False
    
    async def run_load_test(self):
        """执行负载测试"""
        connector = aiohttp.TCPConnector(limit=self.concurrent_users)
        timeout = aiohttp.ClientTimeout(total=30)
        
        async with aiohttp.ClientSession(
            connector=connector, 
            timeout=timeout
        ) as session:
            tasks = []
            for i in range(self.total_requests):
                user_id = i % self.concurrent_users + 1
                task = asyncio.create_task(self.single_request(session, user_id))
                tasks.append(task)
                
                # 控制并发度
                if len(tasks) >= self.concurrent_users:
                    await asyncio.gather(*tasks)
                    tasks = []
            
            if tasks:
                await asyncio.gather(*tasks)
    
    def analyze_results(self):
        """分析测试结果"""
        if not self.response_times:
            return None
            
        metrics = {
            'total_requests': self.total_requests,
            'successful_requests': len(self.response_times),
            'failed_requests': self.error_count,
            'error_rate': (self.error_count / self.total_requests) * 100,
            'response_time': {
                'min': min(self.response_times),
                'max': max(self.response_times),
                'avg': statistics.mean(self.response_times),
                'p50': statistics.median(self.response_times),
                'p90': statistics.quantiles(self.response_times, n=10)[8],
                'p95': statistics.quantiles(self.response_times, n=20)[18],
                'p99': statistics.quantiles(self.response_times, n=100)[98]
            }
        }
        return metrics
 
# TRAE IDE 智能提示：使用 if __name__ == "__main__" 保护主程序入口
if __name__ == "__main__":
    tester = APIPerformanceTester("http://localhost:8080", concurrent_users=50, total_requests=500)
    
    print("开始执行性能测试...")
    start_time = time.time()
    
    # 执行异步测试
    asyncio.run(tester.run_load_test())
    
    duration = time.time() - start_time
    print(f"测试执行时间: {duration:.2f}秒")
    
    # 分析结果
    results = tester.analyze_results()
    print("\n=== 性能测试结果 ===")
    print(f"总请求数: {results['total_requests']}")
    print(f"成功请求: {results['successful_requests']}")
    print(f"失败请求: {results['failed_requests']}")
    print(f"错误率: {results['error_rate']:.2f}%")
    print(f"平均响应时间: {results['response_time']['avg']:.2f}ms")
    print(f"P95响应时间: {results['response_time']['p95']:.2f}ms")

# 使用TRAE IDE调试性能测试脚本
import logging
import cProfile
import pstats
 
# 配置日志 - TRAE IDE会自动识别并高亮日志配置
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
 
logger = logging.getLogger(__name__)
 
def profile_performance_test():
    """性能分析函数 - TRAE IDE支持性能分析可视化"""
    profiler = cProfile.Profile()
    profiler.enable()
    
    # 这里放置需要分析的代码
    tester = APIPerformanceTester("http://localhost:8080")
    asyncio.run(tester.run_load_test())
    
    profiler.disable()
    
    # 输出性能分析结果
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumulative')
    stats.print_stats(20)  # 显示前20个最耗时的函数
 
if __name__ == "__main__":
    profile_performance_test()

# distributed_performance_test.py
import redis
import celery
from celery import Celery
import json
 
# 配置Celery - TRAE IDE会自动提示Celery配置项
app = Celery('performance_test', broker='redis://localhost:6379/0')
 
@app.task
def execute_test_task(test_config):
    """分布式测试任务"""
    tester = APIPerformanceTester(
        test_config['base_url'],
        concurrent_users=test_config['concurrent_users'],
        total_requests=test_config['requests_per_worker']
    )
    
    # 执行测试
    asyncio.run(tester.run_load_test())
    results = tester.analyze_results()
    
    # 存储结果到Redis
    redis_client = redis.Redis(host='localhost', port=6379, db=0)
    redis_client.lpush('test_results', json.dumps(results))
    
    return results
 
class DistributedPerformanceTestManager:
    def __init__(self, worker_count=4):
        self.worker_count = worker_count
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        
    def run_distributed_test(self, test_config):
        """运行分布式测试"""
        # 清空之前的结果
        self.redis_client.delete('test_results')
        
        # 分发任务到多个worker
        tasks = []
        for i in range(self.worker_count):
            task = execute_test_task.delay(test_config)
            tasks.append(task)
        
        # 等待所有任务完成
        results = []
        for task in tasks:
            result = task.get()
            results.append(result)
        
        # 聚合结果
        aggregated_results = self.aggregate_results(results)
        return aggregated_results
    
    def aggregate_results(self, results):
        """聚合多个worker的测试结果"""
        total_requests = sum(r['total_requests'] for r in results)
        total_successful = sum(r['successful_requests'] for r in results)
        total_errors = sum(r['failed_requests'] for r in results)
        
        # 合并所有响应时间
        all_response_times = []
        for result in results:
            all_response_times.extend(result.get('response_times', []))
        
        aggregated = {
            'total_requests': total_requests,
            'successful_requests': total_successful,
            'failed_requests': total_errors,
            'error_rate': (total_errors / total_requests) * 100 if total_requests > 0 else 0,
            'throughput_rps': total_successful / self.calculate_test_duration(),
            'response_time_analysis': self.analyze_response_times(all_response_times)
        }
        
        return aggregated

# performance_visualization.py
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
from datetime import datetime
 
class PerformanceVisualizer:
    def __init__(self, results):
        self.results = results
        self.setup_style()
    
    def setup_style(self):
        """设置图表样式 - TRAE IDE支持matplotlib代码补全"""
        plt.style.use('seaborn-v0_8')
        sns.set_palette("husl")
    
    def plot_response_time_distribution(self):
        """响应时间分布图"""
        response_times = self.results['response_times']
        
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
        
        # 直方图
        ax1.hist(response_times, bins=50, alpha=0.7, color='skyblue', edgecolor='black')
        ax1.set_xlabel('响应时间 (ms)')
        ax1.set_ylabel('频次')
        ax1.set_title('响应时间分布直方图')
        ax1.grid(True, alpha=0.3)
        
        # 箱线图
        ax2.boxplot(response_times, vert=True)
        ax2.set_ylabel('响应时间 (ms)')
        ax2.set_title('响应时间箱线图')
        ax2.grid(True, alpha=0.3)
        
        plt.tight_layout()
        plt.savefig(f'response_time_distribution_{datetime.now().strftime("%Y%m%d_%H%M%S")}.png', dpi=300, bbox_inches='tight')
        plt.show()
    
    def plot_throughput_over_time(self, time_window=10):
        """吞吐量随时间变化图"""
        # 假设我们有时间戳数据
        timestamps = self.results.get('timestamps', [])
        
        if not timestamps:
            print("没有时间戳数据，无法生成吞吐量趋势图")
            return
        
        # 按时间窗口聚合
        df = pd.DataFrame({
            'timestamp': pd.to_datetime(timestamps, unit='s'),
            'requests': 1
        })
        
        # 重采样计算吞吐量
        throughput = df.set_index('timestamp').resample(f'{time_window}s').sum()
        
        plt.figure(figsize=(12, 6))
        plt.plot(throughput.index, throughput['requests'], marker='o', linewidth=2)
        plt.xlabel('时间')
        plt.ylabel(f'{time_window}秒内请求数')
        plt.title('吞吐量时间趋势图')
        plt.grid(True, alpha=0.3)
        plt.xticks(rotation=45)
        
        plt.tight_layout()
        plt.savefig(f'throughput_trend_{datetime.now().strftime("%Y%m%d_%H%M%S")}.png', dpi=300, bbox_inches='tight')
        plt.show()
 
# TRAE IDE 会自动识别matplotlib相关导入并提供代码建议
if __name__ == "__main__":
    # 使用示例
    visualizer = PerformanceVisualizer(test_results)
    visualizer.plot_response_time_distribution()
    visualizer.plot_throughput_over_time()

# optimized_performance_test.py
import asyncio
import aiohttp
import time
import statistics
from typing import List, Dict, Optional
import logging
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
import resource
 
# TRAE IDE 会提示添加类型注解以提高代码质量
@dataclass
class TestConfig:
    base_url: str
    concurrent_users: int
    total_requests: int
    request_timeout: int = 30
    retry_attempts: int = 3
    retry_delay: float = 1.0
 
class OptimizedPerformanceTester:
    """优化后的性能测试器"""
    
    def __init__(self, config: TestConfig):
        self.config = config
        self.response_times: List[float] = []
        self.errors: List[Dict] = []
        self.success_count = 0
        self.start_time: Optional[float] = None
        
        # 配置日志 - TRAE IDE支持日志配置智能提示
        self.logger = logging.getLogger(__name__)
        handler = logging.StreamHandler()
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
    
    async def create_session(self) -> aiohttp.ClientSession:
        """创建优化的HTTP会话"""
        connector = aiohttp.TCPConnector(
            limit=self.config.concurrent_users,
            limit_per_host=self.config.concurrent_users,
            ttl_dns_cache=300,
            use_dns_cache=True,
            keepalive_timeout=30
        )
        
        timeout = aiohttp.ClientTimeout(
            total=self.config.request_timeout,
            connect=10,
            sock_read=self.config.request_timeout - 10
        )
        
        return aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
            headers={'Connection': 'keep-alive'}
        )
    
    async def execute_request_with_retry(self, session: aiohttp.ClientSession, 
                                       user_id: int) -> bool:
        """执行带重试机制的请求"""
        for attempt in range(self.config.retry_attempts):
            try:
                start_time = time.time()
                
                async with session.get(
                    f"{self.config.base_url}/api/users/{user_id}",
                    ssl=False  # 测试环境使用
                ) as response:
                    await response.text()  # 确保响应完整读取
                    
                    response_time = (time.time() - start_time) * 1000
                    self.response_times.append(response_time)
                    
                    if response.status == 200:
                        self.success_count += 1
                        return True
                    else:
                        self._record_error(user_id, response.status, 
                                         "Non-200 status code")
                        
            except asyncio.TimeoutError:
                self._record_error(user_id, -1, "Request timeout")
            except aiohttp.ClientError as e:
                self._record_error(user_id, -1, f"Client error: {str(e)}")
            except Exception as e:
                self._record_error(user_id, -1, f"Unexpected error: {str(e)}")
            
            if attempt < self.config.retry_attempts - 1:
                await asyncio.sleep(self.config.retry_delay * (2 ** attempt))
        
        return False
    
    def _record_error(self, user_id: int, status_code: int, error_msg: str):
        """记录错误信息"""
        self.errors.append({
            'user_id': user_id,
            'status_code': status_code,
            'error': error_msg,
            'timestamp': time.time()
        })
    
    async def run_performance_test(self):
        """执行性能测试主流程"""
        self.start_time = time.time()
        self.logger.info(f"开始性能测试 - 并发用户数: {self.config.concurrent_users}, "
                        f"总请求数: {self.config.total_requests}")
        
        async with await self.create_session() as session:
            # 使用信号量控制并发度
            semaphore = asyncio.Semaphore(self.config.concurrent_users)
            
            async def bounded_request(user_id: int):
                async with semaphore:
                    return await self.execute_request_with_retry(session, user_id)
            
            # 创建任务列表
            tasks = []
            for i in range(self.config.total_requests):
                user_id = i % self.config.concurrent_users + 1
                task = asyncio.create_task(bounded_request(user_id))
                tasks.append(task)
            
            # 批量执行并等待完成
            await asyncio.gather(*tasks, return_exceptions=True)
        
        duration = time.time() - self.start_time
        self.logger.info(f"性能测试完成 - 总耗时: {duration:.2f}秒")
    
    def generate_performance_report(self) -> Dict:
        """生成性能测试报告"""
        if not self.response_times:
            return {'error': 'No successful requests recorded'}
        
        total_requests = self.config.total_requests
        failed_requests = len(self.errors)
        success_rate = (self.success_count / total_requests) * 100
        
        # 计算百分位数
        sorted_times = sorted(self.response_times)
        p50_idx = int(len(sorted_times) * 0.5)
        p90_idx = int(len(sorted_times) * 0.9)
        p95_idx = int(len(sorted_times) * 0.95)
        p99_idx = int(len(sorted_times) * 0.99)
        
        report = {
            'summary': {
                'total_requests': total_requests,
                'successful_requests': self.success_count,
                'failed_requests': failed_requests,
                'success_rate': success_rate,
                'test_duration': time.time() - self.start_time if self.start_time else 0
            },
            'response_time_metrics': {
                'min_ms': min(self.response_times),
                'max_ms': max(self.response_times),
                'mean_ms': statistics.mean(self.response_times),
                'median_ms': statistics.median(self.response_times),
                'std_dev_ms': statistics.stdev(self.response_times),
                'p50_ms': sorted_times[p50_idx],
                'p90_ms': sorted_times[p90_idx],
                'p95_ms': sorted_times[p95_idx],
                'p99_ms': sorted_times[p99_idx]
            },
            'throughput': {
                'requests_per_second': self.success_count / (time.time() - self.start_time) 
                                   if self.start_time else 0
            },
            'error_analysis': {
                'total_errors': failed_requests,
                'error_types': self._categorize_errors()
            }
        }
        
        return report
    
    def _categorize_errors(self) -> Dict[str, int]:
        """错误分类统计"""
        error_categories = {}
        for error in self.errors:
            error_type = error['error'].split(':')[0]
            error_categories[error_type] = error_categories.get(error_type, 0) + 1
        return error_categories
 
# TRAE IDE 智能提示最佳实践
if __name__ == "__main__":
    # 配置测试参数
    config = TestConfig(
        base_url="http://localhost:8080",
        concurrent_users=100,
        total_requests=1000,
        request_timeout=30
    )
    
    # 创建测试器并执行测试
    tester = OptimizedPerformanceTester(config)
    
    # TRAE IDE 支持异步代码调试
    asyncio.run(tester.run_performance_test())
    
    # 生成报告
    report = tester.generate_performance_report()
    
    # 输出结果
    print("\n" + "="*50)
    print("性能测试报告")
    print("="*50)
    print(f"成功率: {report['summary']['success_rate']:.2f}%")
    print(f"平均响应时间: {report['response_time_metrics']['mean_ms']:.2f}ms")
    print(f"P95响应时间: {report['response_time_metrics']['p95_ms']:.2f}ms")
    print(f"吞吐量: {report['throughput']['requests_per_second']:.2f} RPS")

# .github/workflows/performance-test.yml
name: 性能测试自动化
 
on:
  pull_request:
    branches: [ main, develop ]
  schedule:
    - cron: '0 2 * * *'  # 每天凌晨2点执行
 
jobs:
  performance-test:
    runs-on: ubuntu-latest
    
    services:
      redis:
        image: redis:6-alpine
        ports:
          - 6379:6379
      
      postgres:
        image: postgres:13
        env:
          POSTGRES_PASSWORD: testpass
          POSTGRES_DB: testdb
        ports:
          - 5432:5432
    
    steps:
    - uses: actions/checkout@v3
    
    - name: 设置Python环境
      uses: actions/setup-python@v4
      with:
        python-version: '3.9'
    
    - name: 安装依赖
      run: |
        pip install -r requirements.txt
        pip install pytest pytest-asyncio pytest-html
    
    - name: 启动测试服务
      run: |
        # 启动被测试的应用服务
        docker-compose -f docker-compose.test.yml up -d
        sleep 30  # 等待服务完全启动
    
    - name: 执行性能测试
      run: |
        python -m pytest tests/performance/ \
          --html=performance_report.html \
          --self-contained-html \
          --tb=short
    
    - name: 上传测试报告
      uses: actions/upload-artifact@v3
      if: always()
      with:
        name: performance-report
        path: performance_report.html
    
    - name: 性能阈值检查
      run: |
        python scripts/check_performance_thresholds.py \
          --report performance_report.html \
          --thresholds config/performance_thresholds.json