1099 lines
42 KiB
Python
Executable File
1099 lines
42 KiB
Python
Executable File
"""
|
||
Website Stress Testing Tool
|
||
===========================
|
||
纯 Python 异步压力测试工具，基于 aiohttp 实现高并发请求。
|
||
|
||
功能：
|
||
- 并发用户模拟（控制并发数/总请求数/持续时间）
|
||
- 多 HTTP 方法支持（GET/POST/PUT/DELETE/PATCH/HEAD）
|
||
- 自定义请求头、请求体、Cookie
|
||
- 实时终端统计面板（QPS、延迟、成功率）
|
||
- 详细最终报告（百分位延迟、错误分布、吞吐量图表）
|
||
- 自动探测临界并发点并持续维持压测
|
||
- 极限测试模式：探测临界点、警告点、崩溃点
|
||
|
||
使用示例：
|
||
# 基础 GET 测试：100 并发，共 1000 请求
|
||
python stress_test.py -u https://example.com -c 100 -n 1000
|
||
|
||
# 持续 30 秒测试，50 并发
|
||
python stress_test.py -u https://example.com -c 50 -d 30
|
||
|
||
# POST 请求带 JSON body
|
||
python stress_test.py -u https://api.example.com/login -m POST \
|
||
--data '{"user":"test","pass":"123"}' \
|
||
-H "Content-Type: application/json" -c 20 -n 200
|
||
|
||
# 从文件读取 URL 列表轮询测试
|
||
python stress_test.py -f urls.txt -c 50 -d 60
|
||
|
||
# 自动探测临界点，然后持续 2 小时
|
||
python stress_test.py -u https://example.com --find-limit --hours 2
|
||
|
||
# 极限测试：探测临界点、警告点、崩溃点
|
||
python stress_test.py -u https://example.com --crash-test
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
import statistics
|
||
import warnings
|
||
from collections import defaultdict
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timedelta
|
||
from typing import Optional, List, Dict
|
||
|
||
try:
|
||
import aiohttp
|
||
except ImportError:
|
||
print("需要安装 aiohttp: pip install aiohttp")
|
||
sys.exit(1)
|
||
|
||
|
||
# --------------------------- ANSI 颜色支持 ---------------------------
|
||
|
||
_NO_COLOR = False
|
||
|
||
def _supports_color() -> bool:
|
||
"""检测终端是否支持 ANSI 颜色"""
|
||
if os.environ.get("NO_COLOR"):
|
||
return False
|
||
if not hasattr(sys.stdout, "isatty"):
|
||
return False
|
||
return sys.stdout.isatty()
|
||
|
||
def c(text: str, code: str) -> str:
|
||
"""条件着色：仅在支持颜色时添加 ANSI 转义"""
|
||
if _NO_COLOR:
|
||
return text
|
||
return f"\033[{code}m{text}\033[0m"
|
||
|
||
def cbold(text: str) -> str:
|
||
return c(text, "1")
|
||
|
||
def ccyan(text: str) -> str:
|
||
return c(text, "1;36")
|
||
|
||
def cyellow(text: str) -> str:
|
||
return c(text, "1;33")
|
||
|
||
def cgreen(text: str) -> str:
|
||
return c(text, "1;32")
|
||
|
||
def cred(text: str) -> str:
|
||
return c(text, "1;31")
|
||
|
||
def cmagenta(text: str) -> str:
|
||
return c(text, "1;35")
|
||
|
||
def cdim(text: str) -> str:
|
||
return c(text, "32") # 暗绿/灰色，用于进度条
|
||
|
||
def cdim_yellow(text: str) -> str:
|
||
return c(text, "33")
|
||
|
||
def cdim_red(text: str) -> str:
|
||
return c(text, "31")
|
||
|
||
|
||
def format_bytes(n: float) -> str:
|
||
"""格式化字节数为人类可读字符串"""
|
||
for unit in ["B", "KB", "MB", "GB"]:
|
||
if abs(n) < 1024:
|
||
return f"{n:.1f} {unit}"
|
||
n /= 1024
|
||
return f"{n:.1f} TB"
|
||
|
||
|
||
# --------------------------- 数据结构 ---------------------------
|
||
|
||
@dataclass
|
||
class RequestResult:
|
||
"""单次请求结果"""
|
||
status: int # HTTP 状态码，0 表示连接失败
|
||
elapsed: float # 耗时（秒）
|
||
size: int # 响应体大小（字节）
|
||
error: str = "" # 错误信息
|
||
timestamp: float = 0 # 请求发起时间戳
|
||
|
||
|
||
@dataclass
|
||
class TestConfig:
|
||
"""测试配置"""
|
||
urls: list[str]
|
||
method: str = "GET"
|
||
headers: dict = field(default_factory=dict)
|
||
cookies: dict = field(default_factory=dict)
|
||
data: Optional[str] = None
|
||
json_data: Optional[dict] = None
|
||
concurrency: int = 10
|
||
total_requests: Optional[int] = None
|
||
duration: Optional[int] = None # 秒
|
||
timeout: int = 30
|
||
ramp_up: int = 0 # 渐进加压时间（秒）
|
||
verify_ssl: bool = True
|
||
keep_alive: bool = True
|
||
no_display: bool = False
|
||
# find-limit 模式
|
||
find_limit: bool = False
|
||
hours: float = 0.0 # 临界点后持续小时数
|
||
step: int = 10 # 并发递增步长
|
||
probe_time: int = 10 # 每级探测持续秒数
|
||
threshold: float = 95.0 # 成功率临界阈值百分比
|
||
# crash-test 模式
|
||
crash_test: bool = False
|
||
crash_threshold: float = 10.0 # 崩溃阈值：成功率低于此值视为崩溃
|
||
|
||
|
||
# --------------------------- 统计收集器 ---------------------------
|
||
|
||
class StatsCollector:
|
||
"""异步安全的统计数据收集器（仅在单事件循环内使用）"""
|
||
|
||
def __init__(self):
|
||
self.results: list[RequestResult] = []
|
||
self.start_time: float = 0
|
||
self._lock = asyncio.Lock()
|
||
self._status_counts: dict[int, int] = defaultdict(int)
|
||
self._error_counts: dict[str, int] = defaultdict(int)
|
||
self._total_bytes: int = 0
|
||
self._latencies: list[float] = []
|
||
self._completed: int = 0 # 原子计数器，用于控制总请求数
|
||
|
||
async def add(self, result: RequestResult):
|
||
async with self._lock:
|
||
self.results.append(result)
|
||
self._completed += 1
|
||
self._status_counts[result.status] += 1
|
||
self._total_bytes += result.size
|
||
self._latencies.append(result.elapsed)
|
||
if result.error:
|
||
self._error_counts[result.error] += 1
|
||
|
||
async def completed_count(self) -> int:
|
||
async with self._lock:
|
||
return self._completed
|
||
|
||
def total_requests(self) -> int:
|
||
return len(self.results)
|
||
|
||
def success_count(self) -> int:
|
||
return sum(1 for r in self.results if 200 <= r.status < 400)
|
||
|
||
def error_count(self) -> int:
|
||
return sum(1 for r in self.results if r.status >= 400 or r.error)
|
||
|
||
def avg_latency(self) -> float:
|
||
return statistics.mean(self._latencies) if self._latencies else 0.0
|
||
|
||
def percentile(self, p: float) -> float:
|
||
"""计算百分位延迟（线性插值）"""
|
||
if not self._latencies:
|
||
return 0.0
|
||
sorted_lats = sorted(self._latencies)
|
||
n = len(sorted_lats)
|
||
k = (n - 1) * p / 100
|
||
f = int(k)
|
||
c = min(f + 1, n - 1)
|
||
d = k - f
|
||
return sorted_lats[f] + d * (sorted_lats[c] - sorted_lats[f])
|
||
|
||
def elapsed_time(self) -> float:
|
||
return time.time() - self.start_time if self.start_time else 0.0
|
||
|
||
def throughput(self) -> float:
|
||
"""吞吐量 (bytes/sec)"""
|
||
elapsed = self.elapsed_time()
|
||
return self._total_bytes / elapsed if elapsed > 0 else 0.0
|
||
|
||
|
||
# --------------------------- 实时显示 ---------------------------
|
||
|
||
class RealtimeDisplay:
|
||
"""终端实时统计面板"""
|
||
|
||
SPINNER = ["|", "/", "-", "\\"]
|
||
|
||
def __init__(self, stats: StatsCollector, config: TestConfig):
|
||
self.stats = stats
|
||
self.config = config
|
||
self._frame = 0
|
||
self._running = False
|
||
self.critical_concurrency: int = 0 # find-limit 模式下探测到的临界并发数
|
||
|
||
@staticmethod
|
||
def _format_duration(seconds: float) -> str:
|
||
return str(timedelta(seconds=int(seconds)))
|
||
|
||
@staticmethod
|
||
def _progress_bar(current: int, total: int, width: int = 30) -> str:
|
||
if total <= 0:
|
||
return "∞"
|
||
filled = int(width * current / total)
|
||
bar = "#" * filled + "." * (width - filled)
|
||
pct = current / total * 100
|
||
return f"[{bar}] {pct:.1f}%"
|
||
|
||
def render(self) -> str:
|
||
s = self.stats
|
||
total = self.config.total_requests or 0
|
||
duration = self.config.duration or 0
|
||
|
||
spinner = self.SPINNER[self._frame % len(self.SPINNER)]
|
||
self._frame += 1
|
||
|
||
elapsed = s.elapsed_time()
|
||
done = s.total_requests()
|
||
qps = done / elapsed if elapsed > 0 else 0
|
||
|
||
lines = []
|
||
lines.append(ccyan("=" * 60))
|
||
lines.append(f" {spinner} {cbold('Stress Test Running')}")
|
||
lines.append(ccyan("-" * 60))
|
||
|
||
# 进度
|
||
if total > 0:
|
||
lines.append(f" 进度: {self._progress_bar(done, total)} ({done}/{total})")
|
||
elif duration > 0:
|
||
lines.append(f" 进度: {self._progress_bar(int(elapsed), duration)} "
|
||
f"({self._format_duration(elapsed)}/{self._format_duration(duration)})")
|
||
|
||
lines.append("")
|
||
|
||
# 核心指标
|
||
lines.append(f" {cyellow('QPS:')} {qps:>10.1f} req/s")
|
||
lines.append(f" {cyellow('并发数:')} {self.config.concurrency:>10}")
|
||
if self.config.find_limit and self.critical_concurrency > 0:
|
||
remaining = max(0, self.config.duration - elapsed)
|
||
lines.append(f" {cmagenta('临界并发:')} {self.critical_concurrency:>10}")
|
||
lines.append(f" {cmagenta('剩余时间:')} {self._format_duration(remaining):>10}")
|
||
lines.append(f" {cyellow('总请求:')} {done:>10}")
|
||
lines.append(f" {cyellow('耗时:')} {self._format_duration(elapsed):>10}")
|
||
lines.append("")
|
||
|
||
# 延迟
|
||
lines.append(f" {cgreen('平均延迟:')} {s.avg_latency() * 1000:>10.1f} ms")
|
||
lines.append(f" {cgreen('P50:')} {s.percentile(50) * 1000:>10.1f} ms")
|
||
lines.append(f" {cgreen('P90:')} {s.percentile(90) * 1000:>10.1f} ms")
|
||
lines.append(f" {cgreen('P99:')} {s.percentile(99) * 1000:>10.1f} ms")
|
||
lines.append("")
|
||
|
||
# 成功率
|
||
success = s.success_count()
|
||
errors = s.error_count()
|
||
rate = (success / done * 100) if done > 0 else 0
|
||
rate_str = cgreen if rate >= 99 else cyellow if rate >= 95 else cred
|
||
lines.append(f" 成功率: {rate_str(f'{rate:.2f}%')} "
|
||
f"(成功: {success} 失败: {errors})")
|
||
lines.append(f" 吞吐量: {format_bytes(s.throughput())}/s")
|
||
|
||
lines.append(ccyan("=" * 60))
|
||
|
||
return "\n".join(lines)
|
||
|
||
async def run(self, stop_event: asyncio.Event):
|
||
"""每 0.5 秒刷新一次显示"""
|
||
self._running = True
|
||
while self._running and not stop_event.is_set():
|
||
output = self.render()
|
||
# 移动光标到顶部并覆盖输出
|
||
sys.stdout.write(f"\033[{output.count(chr(10)) + 1}A\033[0J")
|
||
sys.stdout.write(output + "\n")
|
||
sys.stdout.flush()
|
||
await asyncio.sleep(0.5)
|
||
|
||
def stop(self):
|
||
self._running = False
|
||
|
||
|
||
# --------------------------- 压力测试引擎 ---------------------------
|
||
|
||
class StressTester:
|
||
"""核心异步压力测试引擎"""
|
||
|
||
def __init__(self, config: TestConfig):
|
||
self.config = config
|
||
self.stats = StatsCollector()
|
||
self._stop_event = asyncio.Event()
|
||
self._active_tasks: set[asyncio.Task] = set()
|
||
|
||
async def _make_request(self, session: aiohttp.ClientSession, url: str) -> RequestResult:
|
||
"""执行单次 HTTP 请求"""
|
||
start = time.time()
|
||
try:
|
||
kwargs = {
|
||
"headers": self.config.headers or None,
|
||
"timeout": aiohttp.ClientTimeout(total=self.config.timeout),
|
||
"ssl": self.config.verify_ssl,
|
||
}
|
||
if self.config.data:
|
||
kwargs["data"] = self.config.data
|
||
elif self.config.json_data:
|
||
kwargs["json"] = self.config.json_data
|
||
|
||
method = self.config.method.upper()
|
||
async with session.request(method, url, **kwargs) as resp:
|
||
body = await resp.read()
|
||
elapsed = time.time() - start
|
||
return RequestResult(
|
||
status=resp.status,
|
||
elapsed=elapsed,
|
||
size=len(body),
|
||
timestamp=start,
|
||
)
|
||
except asyncio.TimeoutError:
|
||
return RequestResult(
|
||
status=0, elapsed=time.time() - start,
|
||
size=0, error="Timeout", timestamp=start,
|
||
)
|
||
except (ConnectionResetError, ConnectionAbortedError):
|
||
return RequestResult(
|
||
status=0, elapsed=time.time() - start,
|
||
size=0, error="ConnectionReset", timestamp=start,
|
||
)
|
||
except aiohttp.ClientError as e:
|
||
return RequestResult(
|
||
status=0, elapsed=time.time() - start,
|
||
size=0, error=type(e).__name__, timestamp=start,
|
||
)
|
||
except Exception as e:
|
||
return RequestResult(
|
||
status=0, elapsed=time.time() - start,
|
||
size=0, error=str(e)[:80], timestamp=start,
|
||
)
|
||
|
||
async def _worker(self, session: aiohttp.ClientSession, url_iter):
|
||
"""单个并发 worker：不断从 url_iter 取 URL 发请求"""
|
||
while not self._stop_event.is_set():
|
||
# 检查总请求数限制（使用锁内计数器避免超发）
|
||
if self.config.total_requests:
|
||
completed = await self.stats.completed_count()
|
||
if completed >= self.config.total_requests:
|
||
break
|
||
try:
|
||
url = next(url_iter)
|
||
except StopIteration:
|
||
break
|
||
result = await self._make_request(session, url)
|
||
await self.stats.add(result)
|
||
|
||
def _url_generator(self):
|
||
"""无限循环生成 URL"""
|
||
urls = self.config.urls
|
||
idx = 0
|
||
while True:
|
||
yield urls[idx % len(urls)]
|
||
idx += 1
|
||
|
||
async def _ramp_up_scheduler(self) -> int:
|
||
"""渐进加压：逐步返回当前应使用的并发数"""
|
||
if self.config.ramp_up <= 0:
|
||
return self.config.concurrency
|
||
|
||
elapsed = self.stats.elapsed_time()
|
||
if elapsed >= self.config.ramp_up:
|
||
return self.config.concurrency
|
||
|
||
ratio = elapsed / self.config.ramp_up
|
||
target = max(1, int(self.config.concurrency * ratio))
|
||
return target
|
||
|
||
async def _control_loop(self, display: RealtimeDisplay | None, stop_display: asyncio.Event):
|
||
"""控制循环：管理任务生命周期和停止条件"""
|
||
url_iter = iter(self._url_generator())
|
||
connector = aiohttp.TCPConnector(
|
||
limit=self.config.concurrency + 10,
|
||
limit_per_host=self.config.concurrency + 10,
|
||
keepalive_timeout=30 if self.config.keep_alive else 0,
|
||
enable_cleanup_closed=True,
|
||
)
|
||
jar = aiohttp.CookieJar()
|
||
if self.config.cookies:
|
||
for k, v in self.config.cookies.items():
|
||
jar.update_cookies({k: v})
|
||
|
||
async with aiohttp.ClientSession(connector=connector, cookie_jar=jar) as session:
|
||
# 主循环
|
||
while not self._stop_event.is_set():
|
||
# 检查停止条件
|
||
if self.config.total_requests:
|
||
completed = await self.stats.completed_count()
|
||
if completed >= self.config.total_requests:
|
||
break
|
||
if self.config.duration and self.stats.elapsed_time() >= self.config.duration:
|
||
break
|
||
|
||
# 渐进加压
|
||
target = await self._ramp_up_scheduler()
|
||
|
||
# 启动新 worker
|
||
while len(self._active_tasks) < target:
|
||
task = asyncio.create_task(self._worker(session, url_iter))
|
||
self._active_tasks.add(task)
|
||
task.add_done_callback(self._active_tasks.discard)
|
||
|
||
await asyncio.sleep(0.05)
|
||
|
||
# 停止所有 worker
|
||
self._stop_event.set()
|
||
if self._active_tasks:
|
||
await asyncio.gather(*self._active_tasks, return_exceptions=True)
|
||
|
||
stop_display.set()
|
||
|
||
async def _find_critical_point(self) -> int:
|
||
"""探测临界并发点：逐步加压，找到性能拐点"""
|
||
step = self.config.step
|
||
probe_time = self.config.probe_time
|
||
threshold = self.config.threshold
|
||
|
||
print(f"\n\033[1;36m[>>] 阶段一：探测临界并发点\033[0m")
|
||
print(f" 步长: {step}")
|
||
print(f" 探测时间: {probe_time}s/级")
|
||
print(f" 成功率阈值: {threshold}%")
|
||
print()
|
||
|
||
baseline_p99 = None
|
||
current_concurrency = step
|
||
last_good_concurrency = 0
|
||
|
||
while True:
|
||
# 为本级探测创建独立的 stats
|
||
probe_stats = StatsCollector()
|
||
probe_stats.start_time = time.time()
|
||
|
||
url_iter = iter(self._url_generator())
|
||
connector = aiohttp.TCPConnector(
|
||
limit=current_concurrency + 10,
|
||
keepalive_timeout=30 if self.config.keep_alive else 0,
|
||
enable_cleanup_closed=True,
|
||
)
|
||
jar = aiohttp.CookieJar()
|
||
if self.config.cookies:
|
||
for k, v in self.config.cookies.items():
|
||
jar.update_cookies({k: v})
|
||
|
||
stop_event = asyncio.Event()
|
||
active_tasks: set[asyncio.Task] = set()
|
||
|
||
async with aiohttp.ClientSession(connector=connector, cookie_jar=jar) as session:
|
||
# 启动 worker
|
||
async def probe_worker():
|
||
while not stop_event.is_set():
|
||
try:
|
||
url = next(url_iter)
|
||
except StopIteration:
|
||
break
|
||
result = await self._make_request(session, url)
|
||
await probe_stats.add(result)
|
||
|
||
for _ in range(current_concurrency):
|
||
task = asyncio.create_task(probe_worker())
|
||
active_tasks.add(task)
|
||
task.add_done_callback(active_tasks.discard)
|
||
|
||
# 等待探测时间
|
||
await asyncio.sleep(probe_time)
|
||
|
||
# 停止 worker
|
||
stop_event.set()
|
||
if active_tasks:
|
||
await asyncio.gather(*active_tasks, return_exceptions=True)
|
||
|
||
# 分析结果
|
||
total = probe_stats.total_requests()
|
||
if total == 0:
|
||
print(f" \033[33m并发 {current_concurrency}: 无请求完成，跳过\033[0m")
|
||
current_concurrency += step
|
||
continue
|
||
|
||
success = probe_stats.success_count()
|
||
rate = (success / total * 100) if total > 0 else 0
|
||
p99 = probe_stats.percentile(99) * 1000 # 转为 ms
|
||
|
||
if baseline_p99 is None:
|
||
baseline_p99 = p99 if p99 > 0 else 1.0
|
||
|
||
# 打印本级结果
|
||
rate_color = "\033[32m" if rate >= threshold else "\033[31m"
|
||
print(f" 并发 {current_concurrency:>5}: "
|
||
f"成功率 {rate_color}{rate:.1f}%\033[0m "
|
||
f"P99 {p99:.1f}ms "
|
||
f"({total} 请求)")
|
||
|
||
# 判断是否到达临界点
|
||
reached_limit = False
|
||
if rate < threshold:
|
||
print(f" \033[1;31m[!] 成功率 {rate:.1f}% < {threshold}%，到达临界点\033[0m")
|
||
reached_limit = True
|
||
elif baseline_p99 > 0 and p99 > baseline_p99 * 5:
|
||
print(f" \033[1;31m[!] P99 {p99:.1f}ms > 基线 {baseline_p99:.1f}ms × 5，到达临界点\033[0m")
|
||
reached_limit = True
|
||
|
||
if reached_limit:
|
||
critical = last_good_concurrency
|
||
if critical == 0:
|
||
critical = max(1, current_concurrency - step)
|
||
print(f"\n \033[1;32m[*] 临界并发数: {critical}\033[0m")
|
||
return critical
|
||
|
||
# 本级通过，记录为"好的"并发数
|
||
last_good_concurrency = current_concurrency
|
||
current_concurrency += step
|
||
|
||
async def _find_crash_point(self) -> dict:
|
||
"""极限测试：探测临界点、警告点、崩溃点"""
|
||
step = self.config.step
|
||
probe_time = self.config.probe_time
|
||
threshold = self.config.threshold
|
||
crash_threshold = self.config.crash_threshold
|
||
|
||
print(f"\n\033[1;36m[>>] 极限测试模式\033[0m")
|
||
print(f" 步长: {step}")
|
||
print(f" 探测时间: {probe_time}s/级")
|
||
print(f" 临界阈值: {threshold}%")
|
||
print(f" 崩溃阈值: {crash_threshold}%")
|
||
print()
|
||
|
||
baseline_p99 = None
|
||
current_concurrency = step
|
||
last_good_concurrency = 0
|
||
critical_concurrency = 0
|
||
warning_concurrency = 0
|
||
crash_concurrency = 0
|
||
|
||
# Phase 1 & 2: 逐步加压直到崩溃
|
||
while True:
|
||
probe_stats = StatsCollector()
|
||
probe_stats.start_time = time.time()
|
||
|
||
url_iter = iter(self._url_generator())
|
||
connector = aiohttp.TCPConnector(
|
||
limit=current_concurrency + 10,
|
||
keepalive_timeout=30 if self.config.keep_alive else 0,
|
||
enable_cleanup_closed=True,
|
||
)
|
||
jar = aiohttp.CookieJar()
|
||
if self.config.cookies:
|
||
for k, v in self.config.cookies.items():
|
||
jar.update_cookies({k, v})
|
||
|
||
stop_event = asyncio.Event()
|
||
active_tasks: set[asyncio.Task] = set()
|
||
|
||
async with aiohttp.ClientSession(connector=connector, cookie_jar=jar) as session:
|
||
async def probe_worker():
|
||
while not stop_event.is_set():
|
||
try:
|
||
url = next(url_iter)
|
||
except StopIteration:
|
||
break
|
||
result = await self._make_request(session, url)
|
||
await probe_stats.add(result)
|
||
|
||
for _ in range(current_concurrency):
|
||
task = asyncio.create_task(probe_worker())
|
||
active_tasks.add(task)
|
||
task.add_done_callback(active_tasks.discard)
|
||
|
||
await asyncio.sleep(probe_time)
|
||
|
||
stop_event.set()
|
||
if active_tasks:
|
||
await asyncio.gather(*active_tasks, return_exceptions=True)
|
||
|
||
# 分析结果
|
||
total = probe_stats.total_requests()
|
||
if total == 0:
|
||
print(f" \033[33m并发 {current_concurrency}: 无请求完成，跳过\033[0m")
|
||
current_concurrency += step
|
||
continue
|
||
|
||
success = probe_stats.success_count()
|
||
rate = (success / total * 100) if total > 0 else 0
|
||
p99 = probe_stats.percentile(99) * 1000
|
||
|
||
if baseline_p99 is None:
|
||
baseline_p99 = p99 if p99 > 0 else 1.0
|
||
|
||
# 打印本级结果
|
||
if rate >= threshold:
|
||
rate_color = "\033[32m"
|
||
elif rate >= crash_threshold:
|
||
rate_color = "\033[33m"
|
||
else:
|
||
rate_color = "\033[31m"
|
||
print(f" 并发 {current_concurrency:>5}: "
|
||
f"成功率 {rate_color}{rate:.1f}%\033[0m "
|
||
f"P99 {p99:.1f}ms "
|
||
f"({total} 请求)")
|
||
|
||
# 判断临界点（首次低于 threshold）
|
||
if critical_concurrency == 0 and rate < threshold:
|
||
critical_concurrency = last_good_concurrency if last_good_concurrency > 0 else max(1, current_concurrency - step)
|
||
print(f" \033[1;33m[!] 临界并发点: {critical_concurrency}\033[0m")
|
||
|
||
# 判断警告点（首次低于 50%）
|
||
if warning_concurrency == 0 and rate < 50.0:
|
||
warning_concurrency = current_concurrency
|
||
print(f" \033[1;33m[!] 警告并发点: {warning_concurrency}\033[0m")
|
||
|
||
# 判断崩溃点（首次低于 crash_threshold）
|
||
if rate < crash_threshold:
|
||
crash_concurrency = current_concurrency
|
||
print(f" \033[1;31m[!] 崩溃并发点: {crash_concurrency}\033[0m")
|
||
break
|
||
|
||
last_good_concurrency = current_concurrency
|
||
current_concurrency += step
|
||
|
||
# 如果从未触发临界点或警告点（不太可能，但处理边界情况）
|
||
if critical_concurrency == 0:
|
||
critical_concurrency = last_good_concurrency if last_good_concurrency > 0 else 1
|
||
if warning_concurrency == 0:
|
||
warning_concurrency = crash_concurrency
|
||
|
||
print(f"\n \033[1;36m[*] 极限测试结果:\033[0m")
|
||
print(f" 临界并发点: {critical_concurrency}")
|
||
print(f" 警告并发点: {warning_concurrency}")
|
||
print(f" 崩溃并发点: {crash_concurrency}")
|
||
|
||
return {
|
||
"critical": critical_concurrency,
|
||
"warning": warning_concurrency,
|
||
"crash": crash_concurrency,
|
||
}
|
||
|
||
async def run(self) -> StatsCollector:
|
||
"""执行压力测试"""
|
||
# find-limit 模式：先探测临界点
|
||
critical_concurrency = 0
|
||
crash_results = None
|
||
if self.config.crash_test:
|
||
crash_results = await self._find_crash_point()
|
||
self.stats.start_time = time.time()
|
||
return self.stats
|
||
elif self.config.find_limit:
|
||
critical_concurrency = await self._find_critical_point()
|
||
self.config.concurrency = critical_concurrency
|
||
self.config.duration = int(self.config.hours * 3600)
|
||
# 重置 stats 用于正式测试
|
||
self.stats = StatsCollector()
|
||
|
||
self.stats.start_time = time.time()
|
||
|
||
print(f"\n\033[1;36m[>>] 压力测试开始\033[0m")
|
||
print(f" 目标: {', '.join(self.config.urls)}")
|
||
print(f" 方法: {self.config.method}")
|
||
if self.config.find_limit:
|
||
print(f" 并发: {self.config.concurrency} (临界点)")
|
||
print(f" 持续: {self.config.duration}s ({self.config.hours} 小时)")
|
||
else:
|
||
print(f" 并发: {self.config.concurrency}")
|
||
if self.config.total_requests:
|
||
print(f" 请求数: {self.config.total_requests}")
|
||
if self.config.duration:
|
||
print(f" 持续: {self.config.duration}s")
|
||
print()
|
||
|
||
stop_display = asyncio.Event()
|
||
|
||
if self.config.no_display:
|
||
# 无显示模式：只运行控制循环
|
||
control_task = asyncio.create_task(self._control_loop(None, stop_display))
|
||
await control_task
|
||
else:
|
||
# 实时显示模式
|
||
display = RealtimeDisplay(self.stats, self.config)
|
||
display.critical_concurrency = critical_concurrency
|
||
for _ in range(20):
|
||
print()
|
||
|
||
display_task = asyncio.create_task(display.run(stop_display))
|
||
control_task = asyncio.create_task(self._control_loop(display, stop_display))
|
||
|
||
await control_task
|
||
display.stop()
|
||
await display_task
|
||
|
||
return self.stats
|
||
|
||
|
||
# --------------------------- 报告生成 ---------------------------
|
||
|
||
class ReportGenerator:
|
||
"""生成详细的最终测试报告"""
|
||
|
||
def __init__(self, stats: StatsCollector, config: TestConfig):
|
||
self.stats = stats
|
||
self.config = config
|
||
|
||
def _bar(self, value: float, max_value: float, width: int = 30) -> str:
|
||
if max_value <= 0:
|
||
return ""
|
||
filled = int(width * min(value / max_value, 1.0))
|
||
return "#" * filled + "." * (width - filled)
|
||
|
||
def generate(self) -> str:
|
||
s = self.stats
|
||
elapsed = s.elapsed_time()
|
||
total = s.total_requests()
|
||
success = s.success_count()
|
||
errors = s.error_count()
|
||
|
||
lines = []
|
||
lines.append("")
|
||
lines.append("\033[1;36m" + "=" * 64 + "\033[0m")
|
||
lines.append(" \033[1;33m[*] 压力测试报告\033[0m")
|
||
lines.append(" " + datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
||
lines.append("\033[1;36m" + "=" * 64 + "\033[0m")
|
||
|
||
# 基本信息
|
||
lines.append("")
|
||
lines.append(" \033[1m> 测试配置\033[0m")
|
||
lines.append(f" 目标 URL: {', '.join(self.config.urls)}")
|
||
lines.append(f" HTTP 方法: {self.config.method}")
|
||
lines.append(f" 并发数: {self.config.concurrency}")
|
||
lines.append(f" 总耗时: {elapsed:.2f}s")
|
||
if self.config.find_limit:
|
||
lines.append("")
|
||
lines.append(" \033[1;35m> 临界点探测结果\033[0m")
|
||
lines.append(f" 临界并发数: {self.config.concurrency}")
|
||
lines.append(f" 持续时间: {self.config.hours} 小时")
|
||
lines.append(f" 探测步长: {self.config.step}")
|
||
lines.append(f" 成功率阈值: {self.config.threshold}%")
|
||
if self.config.crash_test:
|
||
lines.append("")
|
||
lines.append(" \033[1;35m> 极限测试结果\033[0m")
|
||
lines.append(f" 探测步长: {self.config.step}")
|
||
lines.append(f" 探测时间: {self.config.probe_time}s/级")
|
||
lines.append(f" 临界阈值: {self.config.threshold}%")
|
||
lines.append(f" 崩溃阈值: {self.config.crash_threshold}%")
|
||
lines.append("")
|
||
|
||
# 请求统计
|
||
lines.append(" \033[1m> 请求统计\033[0m")
|
||
lines.append(f" 总请求数: {total}")
|
||
rate = (success / total * 100) if total > 0 else 0
|
||
color = "\033[32m" if rate >= 99 else "\033[33m" if rate >= 95 else "\033[31m"
|
||
lines.append(f" 成功: \033[32m{success}\033[0m")
|
||
lines.append(f" 失败: \033[31m{errors}\033[0m")
|
||
lines.append(f" 成功率: {color}{rate:.2f}%\033[0m")
|
||
lines.append(f" 平均 QPS: {total / elapsed:.1f} req/s" if elapsed > 0 else "")
|
||
lines.append("")
|
||
|
||
# 延迟分布
|
||
if s._latencies:
|
||
lines.append(" \033[1m> 延迟分布 (ms)\033[0m")
|
||
percentiles = [50, 75, 90, 95, 99, 99.9]
|
||
for p in percentiles:
|
||
val = s.percentile(p) * 1000
|
||
bar = self._bar(val, s.percentile(99.9) * 1000, 20)
|
||
lines.append(f" P{p:<5} {val:>8.1f} ms {bar}")
|
||
|
||
lines.append("")
|
||
lines.append(f" 最小: {min(s._latencies) * 1000:.1f} ms")
|
||
lines.append(f" 最大: {max(s._latencies) * 1000:.1f} ms")
|
||
lines.append(f" 平均: {s.avg_latency() * 1000:.1f} ms")
|
||
if len(s._latencies) > 1:
|
||
std = statistics.stdev(s._latencies) * 1000
|
||
lines.append(f" 标准差: {std:.1f} ms")
|
||
lines.append("")
|
||
|
||
# 吞吐量
|
||
lines.append(" \033[1m> 吞吐量\033[0m")
|
||
total_bytes = s._total_bytes
|
||
for unit in ["B", "KB", "MB", "GB"]:
|
||
if abs(total_bytes) < 1024:
|
||
lines.append(f" 总传输: {total_bytes:.1f} {unit}")
|
||
break
|
||
total_bytes /= 1024
|
||
|
||
tp = s.throughput()
|
||
for unit in ["B/s", "KB/s", "MB/s", "GB/s"]:
|
||
if abs(tp) < 1024:
|
||
lines.append(f" 吞吐量: {tp:.1f} {unit}")
|
||
break
|
||
tp /= 1024
|
||
lines.append("")
|
||
|
||
# 状态码分布
|
||
if s._status_counts:
|
||
lines.append(" \033[1m> 状态码分布\033[0m")
|
||
max_count = max(s._status_counts.values())
|
||
for code, count in sorted(s._status_counts.items()):
|
||
bar = self._bar(count, max_count, 20)
|
||
color = "\033[32m" if 200 <= code < 300 else "\033[33m" if 300 <= code < 400 else "\033[31m"
|
||
label = f"HTTP {code}" if code > 0 else "连接失败"
|
||
lines.append(f" {color}{label:<12}\033[0m {count:>6} {bar}")
|
||
lines.append("")
|
||
|
||
# 错误详情
|
||
if s._error_counts:
|
||
lines.append(" \033[1;31m> 错误详情\033[0m")
|
||
for err, count in sorted(s._error_counts.items(), key=lambda x: -x[1]):
|
||
lines.append(f" \033[31m{err:<30}\033[0m {count:>6} 次")
|
||
lines.append("")
|
||
|
||
# 评价
|
||
lines.append(" \033[1m> 综合评价\033[0m")
|
||
if rate >= 99.9 and s.percentile(99) < 0.5:
|
||
grade = "\033[1;32m[Excellent] 优秀\033[0m - 网站表现稳定，响应迅速"
|
||
elif rate >= 99 and s.percentile(99) < 1.0:
|
||
grade = "\033[1;32m[Good] 良好\033[0m - 网站表现正常"
|
||
elif rate >= 95:
|
||
grade = "\033[1;33m[Fair] 一般\033[0m - 存在部分超时或错误，建议优化"
|
||
elif rate >= 80:
|
||
grade = "\033[1;33m[Poor] 较差\033[0m - 高负载下错误率偏高"
|
||
else:
|
||
grade = "\033[1;31m[Bad] 很差\033[0m - 网站无法承受当前负载"
|
||
|
||
lines.append(f" {grade}")
|
||
lines.append("")
|
||
lines.append("\033[1;36m" + "=" * 64 + "\033[0m")
|
||
|
||
return "\n".join(lines)
|
||
|
||
def save_json(self, filepath: str):
|
||
"""保存 JSON 格式报告"""
|
||
s = self.stats
|
||
elapsed = s.elapsed_time()
|
||
total = s.total_requests()
|
||
|
||
report = {
|
||
"timestamp": datetime.now().isoformat(),
|
||
"config": {
|
||
"urls": self.config.urls,
|
||
"method": self.config.method,
|
||
"concurrency": self.config.concurrency,
|
||
"total_requests": self.config.total_requests,
|
||
"duration": self.config.duration,
|
||
},
|
||
"find_limit": {
|
||
"enabled": self.config.find_limit,
|
||
"critical_concurrency": self.config.concurrency if self.config.find_limit else None,
|
||
"hours": self.config.hours if self.config.find_limit else None,
|
||
"step": self.config.step if self.config.find_limit else None,
|
||
"threshold": self.config.threshold if self.config.find_limit else None,
|
||
} if self.config.find_limit else None,
|
||
"crash_test": {
|
||
"enabled": self.config.crash_test,
|
||
"step": self.config.step,
|
||
"probe_time": self.config.probe_time,
|
||
"threshold": self.config.threshold,
|
||
"crash_threshold": self.config.crash_threshold,
|
||
} if self.config.crash_test else None,
|
||
"results": {
|
||
"total_requests": total,
|
||
"success": s.success_count(),
|
||
"errors": s.error_count(),
|
||
"success_rate": round(s.success_count() / total * 100, 2) if total else 0,
|
||
"elapsed_seconds": round(elapsed, 2),
|
||
"avg_qps": round(total / elapsed, 1) if elapsed else 0,
|
||
"avg_latency_ms": round(s.avg_latency() * 1000, 2),
|
||
"p50_ms": round(s.percentile(50) * 1000, 2),
|
||
"p90_ms": round(s.percentile(90) * 1000, 2),
|
||
"p95_ms": round(s.percentile(95) * 1000, 2),
|
||
"p99_ms": round(s.percentile(99) * 1000, 2),
|
||
"total_bytes": s._total_bytes,
|
||
"throughput_bytes_sec": round(s.throughput(), 1),
|
||
"status_codes": dict(s._status_counts),
|
||
"errors_detail": dict(s._error_counts),
|
||
},
|
||
}
|
||
|
||
with open(filepath, "w", encoding="utf-8") as f:
|
||
json.dump(report, f, indent=2, ensure_ascii=False)
|
||
print(f"\n JSON 报告已保存: {filepath}")
|
||
|
||
|
||
# --------------------------- CLI 入口 ---------------------------
|
||
|
||
def parse_headers(raw: list[str]) -> dict:
|
||
"""解析 -H "Key: Value" 格式的请求头"""
|
||
headers = {}
|
||
for h in raw:
|
||
if ":" in h:
|
||
k, v = h.split(":", 1)
|
||
headers[k.strip()] = v.strip()
|
||
return headers
|
||
|
||
|
||
def parse_cookies(raw: str) -> dict:
|
||
"""解析 k1=v1;k2=v2 格式的 Cookie"""
|
||
cookies = {}
|
||
for pair in raw.split(";"):
|
||
if "=" in pair:
|
||
k, v = pair.split("=", 1)
|
||
cookies[k.strip()] = v.strip()
|
||
return cookies
|
||
|
||
|
||
def main():
|
||
# Windows 终端 UTF-8 支持
|
||
if sys.platform == "win32":
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
sys.stderr.reconfigure(encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
|
||
# 抑制 event loop 关闭时的无害警告
|
||
warnings.filterwarnings("ignore", message=".*Event loop is closed.*")
|
||
|
||
# Windows: 抑制 asyncio 底层连接重置的噪音回调
|
||
if sys.platform == "win32":
|
||
def _silent_handler(loop, context):
|
||
exc = context.get("exception")
|
||
if isinstance(exc, (ConnectionResetError, ConnectionAbortedError, OSError)):
|
||
return # 静默忽略远端强制断开
|
||
loop.default_exception_handler(context)
|
||
|
||
try:
|
||
asyncio.get_event_loop().set_exception_handler(_silent_handler)
|
||
except RuntimeError:
|
||
pass
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description="Website Stress Testing Tool - 异步高并发压力测试工具",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
示例:
|
||
%(prog)s -u https://example.com -c 100 -n 1000
|
||
%(prog)s -u https://example.com -c 50 -d 30
|
||
%(prog)s -u https://api.example.com/login -m POST --data '{"user":"test"}'
|
||
%(prog)s -f urls.txt -c 50 -d 60 --report report.json
|
||
%(prog)s -u https://example.com --find-limit --hours 2
|
||
%(prog)s -u https://example.com --find-limit --hours 1 --step 20 --probe-time 15
|
||
%(prog)s -u https://example.com --crash-test
|
||
%(prog)s -u https://example.com --crash-test --step 20 --probe-time 15
|
||
""",
|
||
)
|
||
|
||
# 目标
|
||
target = parser.add_mutually_exclusive_group(required=True)
|
||
target.add_argument("-u", "--url", action="append", help="目标 URL（可多次指定）")
|
||
target.add_argument("-f", "--file", help="URL 列表文件（每行一个）")
|
||
|
||
# 测试模式（-n/-d 与 --find-limit 互斥）
|
||
mode = parser.add_mutually_exclusive_group(required=True)
|
||
mode.add_argument("-n", "--num-requests", type=int, help="总请求数")
|
||
mode.add_argument("-d", "--duration", type=int, help="持续时间（秒）")
|
||
mode.add_argument("--find-limit", action="store_true",
|
||
help="自动探测临界并发点（需配合 --hours）")
|
||
mode.add_argument("--crash-test", action="store_true",
|
||
help="极限测试：探测临界点、警告点、崩溃点")
|
||
|
||
# 并发
|
||
parser.add_argument("-c", "--concurrency", type=int, default=10, help="并发数 (默认: 10)")
|
||
parser.add_argument("-r", "--ramp-up", type=int, default=0, help="渐进加压时间（秒）(默认: 0)")
|
||
|
||
# find-limit 模式参数
|
||
parser.add_argument("--hours", type=float, default=0,
|
||
help="临界点后持续压测小时数（配合 --find-limit）")
|
||
parser.add_argument("--step", type=int, default=10,
|
||
help="并发递增步长 (默认: 10)")
|
||
parser.add_argument("--probe-time", type=int, default=10,
|
||
help="每级探测持续秒数 (默认: 10)")
|
||
parser.add_argument("--threshold", type=float, default=95,
|
||
help="成功率临界阈值百分比 (默认: 95)")
|
||
parser.add_argument("--crash-threshold", type=float, default=10,
|
||
help="崩溃阈值：成功率低于此值视为崩溃 (默认: 10)")
|
||
|
||
# HTTP 配置
|
||
parser.add_argument("-m", "--method", default="GET",
|
||
choices=["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD"],
|
||
help="HTTP 方法 (默认: GET)")
|
||
parser.add_argument("-H", "--header", action="append", default=[],
|
||
help="自定义请求头 (-H 'Key: Value')")
|
||
parser.add_argument("--cookie", default="", help="Cookie (k1=v1;k2=v2)")
|
||
parser.add_argument("--data", help="请求体（字符串）")
|
||
parser.add_argument("--json", dest="json_data", help="JSON 请求体")
|
||
|
||
# 网络
|
||
parser.add_argument("-t", "--timeout", type=int, default=30, help="请求超时（秒）(默认: 30)")
|
||
parser.add_argument("--no-verify-ssl", action="store_true", help="跳过 SSL 验证")
|
||
parser.add_argument("--no-keep-alive", action="store_true", help="禁用 Keep-Alive")
|
||
|
||
# 输出
|
||
parser.add_argument("--report", help="保存 JSON 报告到文件")
|
||
parser.add_argument("--no-display", action="store_true", help="禁用实时显示（适合日志采集）")
|
||
|
||
args = parser.parse_args()
|
||
|
||
# find-limit 模式验证
|
||
if args.find_limit and args.hours <= 0:
|
||
parser.error("--find-limit 需要配合 --hours 指定持续小时数（必须 > 0）")
|
||
|
||
# 解析 URL
|
||
urls = []
|
||
if args.url:
|
||
urls = args.url
|
||
elif args.file:
|
||
with open(args.file, "r", encoding="utf-8") as f:
|
||
urls = [line.strip() for line in f if line.strip() and not line.startswith("#")]
|
||
if not urls:
|
||
print(f"错误: 文件 {args.file} 中没有有效 URL")
|
||
sys.exit(1)
|
||
|
||
# 确保 URL 有协议前缀
|
||
urls = [u if u.startswith(("http://", "https://")) else "https://" + u for u in urls]
|
||
|
||
# 解析 JSON body
|
||
json_data = None
|
||
if args.json_data:
|
||
try:
|
||
json_data = json.loads(args.json_data)
|
||
except json.JSONDecodeError:
|
||
print("错误: --json 参数不是有效的 JSON")
|
||
sys.exit(1)
|
||
|
||
config = TestConfig(
|
||
urls=urls,
|
||
method=args.method,
|
||
headers=parse_headers(args.header),
|
||
cookies=parse_cookies(args.cookie) if args.cookie else {},
|
||
data=args.data,
|
||
json_data=json_data,
|
||
concurrency=args.concurrency,
|
||
total_requests=args.num_requests,
|
||
duration=args.duration,
|
||
timeout=args.timeout,
|
||
ramp_up=args.ramp_up,
|
||
verify_ssl=not args.no_verify_ssl,
|
||
keep_alive=not args.no_keep_alive,
|
||
no_display=args.no_display,
|
||
find_limit=args.find_limit,
|
||
hours=args.hours,
|
||
step=args.step,
|
||
probe_time=args.probe_time,
|
||
threshold=args.threshold,
|
||
crash_test=args.crash_test,
|
||
crash_threshold=args.crash_threshold,
|
||
)
|
||
|
||
# 运行测试
|
||
try:
|
||
tester = StressTester(config)
|
||
stats = asyncio.run(tester.run())
|
||
except KeyboardInterrupt:
|
||
print("\n\n\033[33m[!] 测试被用户中断\033[0m")
|
||
stats = tester.stats
|
||
|
||
# 生成报告
|
||
report = ReportGenerator(stats, config)
|
||
print(report.generate())
|
||
|
||
if args.report:
|
||
report.save_json(args.report)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|