并发测试网站

import aiohttp
import asyncio
import time

async def test_url_async(session, url):
    """异步测试单个请求"""
    try:
        start_time = time.time()
        async with session.get(url, timeout=10) as response:
            end_time = time.time()
            return {
                "status": "success",
                "status_code": response.status,
                "time": end_time - start_time
            }
    except Exception as e:
        return {
            "status": "fail",
            "error": str(e),
            "time": time.time() - start_time
        }

async def high_concurrency_async_test(url, concurrency, total_requests):
    """异步高并发测试"""
    print(f"开始异步测试：URL={url}，并发数={concurrency}，总请求数={total_requests}")
    start_time = time.time()
    
    success = 0
    fail = 0
    total_time = 0
    max_time = 0
    min_time = float('inf')
    
    # 控制并发数（通过信号量）
    semaphore = asyncio.Semaphore(concurrency)
    
    async def bounded_test():
        async with semaphore:
            return await test_url_async(session, url)
    
    async with aiohttp.ClientSession() as session:
        # 创建所有任务
        tasks = [bounded_test() for _ in range(total_requests)]
        # 并发执行
        results = await asyncio.gather(*tasks)
    
    # 统计结果
    for result in results:
        if result["status"] == "success":
            success += 1
            total_time += result["time"]
            max_time = max(max_time, result["time"])
            min_time = min(min_time, result["time"])
        else:
            fail += 1
            # print(f"请求失败：{result['error']}")
    
    end_time = time.time()
    total_duration = end_time - start_time
    
    # 打印统计结果（同方案1）
    print("\n测试结果：")
    print(f"总耗时：{total_duration:.2f}秒")
    print(f"总请求数：{total_requests}")
    print(f"成功数：{success}，失败数：{fail}")
    print(f"成功率：{success/total_requests*100:.2f}%")
    if success > 0:
        print(f"平均响应时间：{total_time/success:.4f}秒")
        print(f"最大响应时间：{max_time:.4f}秒")
        print(f"最小响应时间：{min_time:.4f}秒")
    print(f"QPS（每秒请求数）：{total_requests/total_duration:.2f}")

# 测试示例
if __name__ == "__main__":
    target_url = "https://020417.xyz"  # 替换为目标URL
    concurrency = 100  # 并发数（异步可支持更高）
    total_requests = 2000  # 总请求数
    asyncio.run(high_concurrency_async_test(target_url, concurrency, total_requests))