Sublime实现高并发接口性能压测脚本_支持并发用户模拟与响应统计

pip install requests aiohttp

import asyncio
import aiohttp
from collections import Counter

async def fetch(session, url):
    async with session.get(url) as response:
        status = response.status
        return status

async def main(url, total_requests):
    connector = aiohttp.TCPConnector(limit_per_host=100)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [fetch(session, url) for _ in range(total_requests)]
        results = await asyncio.gather(*tasks)
        return results

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    url = "https://example.com/api"
    responses = loop.run_until_complete(main(url, 1000))
    print(Counter(responses))

import time

async def fetch(session, url, stats):
    start = time.time()
    try:
        async with session.get(url, timeout=10) as response:
            elapsed = time.time() - start
            stats.append({
                'status': response.status,
                'time': elapsed
            })
    except Exception as e:
        stats.append({
            'status': 'error',
            'error': str(e),
            'time': time.time() - start
        })

times = [r['time'] for r in stats]
avg_time = sum(times) / len(times)
errors = sum(1 for r in stats if r['status'] != 200 and r['status'] != 'error')
print(f"平均响应时间: {avg_time:.2f}s")
print(f"错误率: {errors / len(stats) * 100:.2f}%")

async def main(url, total_requests, concurrency):
    connector = aiohttp.TCPConnector(limit_per_host=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        for i in range(0, total_requests, concurrency):
            batch = [fetch(session, url) for _ in range(concurrency)]
            await asyncio.gather(*batch)
            await asyncio.sleep(1)  # 每批之间停顿一秒