Function test_concurrent_performance

test_performance.py:99–141

测试并发性能

()

        }

def test_concurrent_performance():
    """测试并发性能"""
    # 动态获取可用源
    test_sources = get_available_sources()
    
    if not test_sources:
        print("❌ 没有找到可用的测试源")
        return
    
    print("🚀 开始API密钥管理器性能测试")
    print(f"📊 配置: {CONCURRENT_REQUESTS}并发, 总共{TOTAL_REQUESTS}次请求")
    print(f"🎯 可用源: {', '.join(test_sources)}")
    print("-" * 60)
    
    # 生成测试任务
    tasks = []
    for i in range(TOTAL_REQUESTS):
        source_name = test_sources[i % len(test_sources)]
        tasks.append((source_name, i))
    
    # 并发执行
    start_time = time.time()
    results = []
    
    with ThreadPoolExecutor(max_workers=CONCURRENT_REQUESTS) as executor:
        future_to_task = {
            executor.submit(test_get_api_key, source_name, request_id): (source_name, request_id)
            for source_name, request_id in tasks
        }
        
        for future in as_completed(future_to_task):
            result = future.result()
            results.append(result)
            
            # 实时显示进度
            if len(results) % 10 == 0:
                print(f"⏳ 已完成: {len(results)}/{TOTAL_REQUESTS}")
    
    end_time = time.time()
    total_time = end_time - start_time
    
    # 分析结果
    analyze_results(results, total_time)

def analyze_results(results: list, total_time: float):
    """分析测试结果"""