基于Python编写用于Web综合性能测试的工具

前言

Web压力测试工具，用于测试Web服务器的极限并发量和带宽。通过模拟正常用户访问服务器，并非服务器返回200就算完成，而是并交互其内容完成之后算一次，通过随机UA+Referer防止触发安全预警。支持多线程并发，测试进度可视化，除了Windows的图形化还支持命令行模式。

命令行模式说明

--url	-u	字符串	✅ 是	无	目标 URL，支持 http/https，若未带协议则自动补全
--concurrency	-c	整数	❌ 否	10	并发线程数
--duration	-d	浮点数	✅ 是	无	持续测试时长（秒）
--timeout	-t	浮点数	❌ 否	5.0	每个请求的超时时间（秒）
--method	-m	字符串	❌ 否	GET	HTTP 方法（如 GET、POST、PUT、DELETE）
--success-codes	无	整数列表	❌ 否	[200]	视为成功的状态码，可指定多个，如 --success-codes 200 201
--headers	无	字符串	❌ 否	None	自定义请求头，格式 'key1:value1,key2:value2'
--data	无	字符串	❌ 否	None	POST 请求体数据
--insecure	无	标志（action）	❌ 否	False（验证SSL）	忽略 SSL 证书验证（即不验证证书）
--random-ua	无	标志（action）	❌ 否	True（开启）	每次请求使用随机 User-Agent（默认开启）
--no-random-ua	无	标志（action）	❌ 否	—	禁用随机 User-Agent（关闭该功能）
--random-referer	无	标志（action）	❌ 否	True（开启）	每次请求使用随机 Referer（默认开启）
--no-random-referer	无	标志（action）	❌ 否	—	禁用随机 Referer（关闭该功能）

代码如下：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Web 压力测试工具（持续时长版）- 随机 UA + 随机 Referer
- 支持 HTTP/HTTPS，自动补全协议
- 多线程并发，TCP 连接复用
- 指定测试时长（秒），持续发送请求
- 每次请求使用随机 User-Agent 和 Referer（可开关）
- 可选忽略 SSL 证书验证
- 实时进度（时间进度 + 请求计数），停止按钮，详细统计
- 命令行模式兼容
"""

import argparse
import time
import sys
import signal
import threading
import random
from concurrent.futures import ThreadPoolExecutor

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# ---------- GUI 依赖 ----------
try:
    import tkinter as tk
    from tkinter import ttk, scrolledtext
    GUI_AVAILABLE = True
except ImportError:
    GUI_AVAILABLE = False

# ---------- 全局停止标志 ----------
stop_flag = False

# ---------- 内置 User-Agent 池 ----------
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/114.0",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/113.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.1 Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
    "Mozilla/5.0 (iPhone; CPU iPhone OS 16_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Mobile/15E148 Safari/604.1",
    "Mozilla/5.0 (iPad; CPU OS 16_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Mobile/15E148 Safari/604.1",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.43",
]

# ---------- 内置 Referer 池（常见来源） ----------
REFERERS = [
    "https://www.google.com/",
    "https://www.baidu.com/",
    "https://www.bing.com/",
    "https://www.sogou.com/",
    "https://www.360.cn/",
    "https://www.so.com/",
    "https://www.zhihu.com/",
    "https://www.weibo.com/",
    "https://www.douyin.com/",
    "https://www.qq.com/",
    "https://www.taobao.com/",
    "https://www.jd.com/",
    "https://www.xiaohongshu.com/",
    "https://www.linkedin.com/",
    "https://www.facebook.com/",
    "https://twitter.com/",
]


def signal_handler(sig, frame):
    global stop_flag
    print("\n收到中断信号，正在停止...")
    stop_flag = True


def normalize_url(url, default_protocol='http'):
    """自动补全协议"""
    url = url.strip()
    if not url:
        return ''
    if url.startswith(('http://', 'https://')):
        return url
    return f"{default_protocol}://{url}"


# ---------- 核心压力测试类 ----------
class StressTester:
    def __init__(self, url, concurrency, duration, timeout=5.0,
                 method='GET', success_codes=None, headers=None, data=None,
                 verify_ssl=True, random_ua=True, random_referer=True):
        """
        :param random_ua: 是否每次请求使用随机 User-Agent
        :param random_referer: 是否每次请求使用随机 Referer（若用户未自定义）
        """
        self.url = url
        self.concurrency = concurrency
        self.duration = duration
        self.timeout = timeout
        self.method = method.upper()
        self.success_codes = success_codes or [200]
        self.base_headers = headers or {}
        self.data = data
        self.verify_ssl = verify_ssl
        self.random_ua = random_ua
        self.random_referer = random_referer
        self.results = []
        self.stop_flag = False

    def set_stop_flag(self):
        self.stop_flag = True

    def worker(self, end_time):
        """单个线程：在结束时间前持续发送请求（完整响应体，测试带宽）"""
        session = requests.Session()
        adapter = HTTPAdapter(
            pool_connections=5,
            pool_maxsize=5,
            max_retries=Retry(total=0, read=0, connect=0)
        )
        session.mount('http://', adapter)
        session.mount('https://', adapter)
        session.verify = self.verify_ssl

        results = []
        while time.time() < end_time and not self.stop_flag:
            start = time.time()
            try:
                # 构建请求头：复制基础 Headers
                headers = self.base_headers.copy()
                if self.random_ua:
                    headers['User-Agent'] = random.choice(USER_AGENTS)
                # 如果用户没有自定义 Referer，且开启了随机 Referer，则添加
                if 'Referer' not in headers and self.random_referer:
                    headers['Referer'] = random.choice(REFERERS)

                # 发送请求（默认 stream=False，下载完整响应体）
                if self.method == 'GET':
                    resp = session.get(self.url, timeout=self.timeout, headers=headers)
                elif self.method == 'POST':
                    resp = session.post(self.url, timeout=self.timeout, headers=headers, data=self.data)
                else:
                    resp = session.request(self.method, self.url, timeout=self.timeout,
                                           headers=headers, data=self.data)
                status = resp.status_code
                elapsed = time.time() - start
                success = status in self.success_codes
                results.append((success, elapsed, status))
            except Exception as e:
                elapsed = time.time() - start
                results.append((False, elapsed, None))
                # 如需调试可打印异常：print(f"请求异常: {e}")
        return results

    def run(self, progress_callback=None, result_callback=None):
        self.stop_flag = False
        start_time = time.time()
        end_time = start_time + self.duration

        all_results = []
        sent_count = 0
        last_update = 0

        with ThreadPoolExecutor(max_workers=self.concurrency) as executor:
            futures = [executor.submit(self.worker, end_time) for _ in range(self.concurrency)]

            while futures:
                done = [f for f in futures if f.done()]
                for f in done:
                    futures.remove(f)
                    try:
                        res = f.result()
                        all_results.extend(res)
                        sent_count += len(res)
                    except Exception:
                        pass

                elapsed = time.time() - start_time
                if progress_callback and (elapsed - last_update >= 0.1 or sent_count % 10 == 0):
                    progress_callback(elapsed, self.duration, sent_count)
                    last_update = elapsed

                if self.stop_flag:
                    for f in futures:
                        f.cancel()
                    break

                if not futures:
                    break

                time.sleep(0.05)

        total_time = time.time() - start_time
        total_requests = len(all_results)

        successes = sum(1 for r in all_results if r[0])
        failures = total_requests - successes
        success_rate = (successes / total_requests * 100) if total_requests > 0 else 0.0
        avg_time_all = sum(r[1] for r in all_results) / total_requests if total_requests > 0 else 0.0
        success_times = [r[1] for r in all_results if r[0]]
        if success_times:
            avg_time_success = sum(success_times) / len(success_times)
            min_time = min(success_times)
            max_time = max(success_times)
        else:
            avg_time_success = min_time = max_time = 0.0
        qps = total_requests / total_time if total_time > 0 else 0.0

        stats = {
            'total_requests': total_requests,
            'successes': successes,
            'failures': failures,
            'success_rate': success_rate,
            'total_time': total_time,
            'qps': qps,
            'avg_time_all': avg_time_all,
            'avg_time_success': avg_time_success,
            'min_time': min_time,
            'max_time': max_time,
        }
        if result_callback:
            result_callback(stats)
        return stats


# ---------- GUI 应用程序 ----------
class StressTestApp:
    def __init__(self, root):
        self.root = root
        self.root.title("Web 压力测试工具 (持续时长 + 随机UA/Referer)")
        self.root.geometry("800x750")
        self.root.resizable(True, True)

        self.tester = None
        self.test_thread = None
        self.running = False

        self.create_widgets()

    def create_widgets(self):
        main_frame = ttk.Frame(self.root, padding="10")
        main_frame.pack(fill=tk.BOTH, expand=True)

        # 第一行：URL + 协议
        row1 = ttk.Frame(main_frame)
        row1.pack(fill=tk.X, pady=2)
        ttk.Label(row1, text="目标 URL:").pack(side=tk.LEFT, padx=(0, 5))
        self.url_var = tk.StringVar(value="example.com")
        url_entry = ttk.Entry(row1, textvariable=self.url_var, width=40)
        url_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 5))
        self.protocol_var = tk.StringVar(value="http")
        protocol_combo = ttk.Combobox(row1, textvariable=self.protocol_var, values=["http", "https"], width=6)
        protocol_combo.pack(side=tk.LEFT)
        protocol_combo.current(0)

        # 第二行：并发数、持续时长、超时
        row2 = ttk.Frame(main_frame)
        row2.pack(fill=tk.X, pady=2)
        ttk.Label(row2, text="并发数:").pack(side=tk.LEFT, padx=(0, 5))
        self.concurrency_var = tk.IntVar(value=10)
        ttk.Spinbox(row2, from_=1, to=1000, textvariable=self.concurrency_var, width=8).pack(side=tk.LEFT, padx=(0, 15))

        ttk.Label(row2, text="持续时长(秒):").pack(side=tk.LEFT, padx=(0, 5))
        self.duration_var = tk.DoubleVar(value=10.0)
        ttk.Spinbox(row2, from_=0.5, to=3600.0, increment=0.5, textvariable=self.duration_var, width=8).pack(side=tk.LEFT, padx=(0, 15))

        ttk.Label(row2, text="超时(s):").pack(side=tk.LEFT, padx=(0, 5))
        self.timeout_var = tk.DoubleVar(value=5.0)
        ttk.Spinbox(row2, from_=0.5, to=60.0, increment=0.5, textvariable=self.timeout_var, width=6).pack(side=tk.LEFT)

        # 第三行：方法、成功状态码、SSL验证、随机UA、随机Referer
        row3 = ttk.Frame(main_frame)
        row3.pack(fill=tk.X, pady=2)
        ttk.Label(row3, text="方法:").pack(side=tk.LEFT, padx=(0, 5))
        self.method_var = tk.StringVar(value="GET")
        method_combo = ttk.Combobox(row3, textvariable=self.method_var, values=["GET", "POST", "PUT", "DELETE"], width=6)
        method_combo.pack(side=tk.LEFT, padx=(0, 15))
        ttk.Label(row3, text="成功状态码:").pack(side=tk.LEFT, padx=(0, 5))
        self.codes_var = tk.StringVar(value="200")
        ttk.Entry(row3, textvariable=self.codes_var, width=15).pack(side=tk.LEFT, padx=(0, 15))
        self.ignore_ssl_var = tk.BooleanVar(value=False)
        ttk.Checkbutton(row3, text="忽略SSL证书验证", variable=self.ignore_ssl_var).pack(side=tk.LEFT, padx=(5, 0))
        self.random_ua_var = tk.BooleanVar(value=True)
        ttk.Checkbutton(row3, text="随机User-Agent", variable=self.random_ua_var).pack(side=tk.LEFT, padx=(5, 0))
        self.random_referer_var = tk.BooleanVar(value=True)
        ttk.Checkbutton(row3, text="随机Referer", variable=self.random_referer_var).pack(side=tk.LEFT, padx=(5, 0))

        # 第四行：Headers
        row4 = ttk.Frame(main_frame)
        row4.pack(fill=tk.X, pady=2)
        ttk.Label(row4, text="Headers (key:val,key2:val2):").pack(side=tk.LEFT, padx=(0, 5))
        self.headers_var = tk.StringVar(value="")
        ttk.Entry(row4, textvariable=self.headers_var, width=40).pack(side=tk.LEFT, fill=tk.X, expand=True)

        # 第五行：POST Data
        row5 = ttk.Frame(main_frame)
        row5.pack(fill=tk.X, pady=2)
        ttk.Label(row5, text="POST Data:").pack(side=tk.LEFT, padx=(0, 5))
        self.data_var = tk.StringVar(value="")
        ttk.Entry(row5, textvariable=self.data_var, width=40).pack(side=tk.LEFT, fill=tk.X, expand=True)

        # 按钮和进度条
        row_btn = ttk.Frame(main_frame)
        row_btn.pack(fill=tk.X, pady=5)
        self.start_btn = ttk.Button(row_btn, text="开始测试", command=self.start_test)
        self.start_btn.pack(side=tk.LEFT, padx=5)
        self.stop_btn = ttk.Button(row_btn, text="停止", command=self.stop_test, state=tk.DISABLED)
        self.stop_btn.pack(side=tk.LEFT, padx=5)

        progress_frame = ttk.Frame(row_btn)
        progress_frame.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=10)
        self.progress_var = tk.IntVar(value=0)
        self.progress_bar = ttk.Progressbar(progress_frame, variable=self.progress_var, maximum=100, length=200)
        self.progress_bar.pack(side=tk.LEFT, fill=tk.X, expand=True)
        self.progress_label = ttk.Label(progress_frame, text="0%")
        self.progress_label.pack(side=tk.LEFT, padx=(5, 10))
        self.req_count_var = tk.StringVar(value="请求: 0")
        req_label = ttk.Label(row_btn, textvariable=self.req_count_var)
        req_label.pack(side=tk.LEFT, padx=5)

        # 日志框
        log_frame = ttk.LabelFrame(main_frame, text="测试日志", padding="5")
        log_frame.pack(fill=tk.BOTH, expand=True, pady=5)
        self.log_text = scrolledtext.ScrolledText(log_frame, height=10, state=tk.NORMAL)
        self.log_text.pack(fill=tk.BOTH, expand=True)

        # 统计结果
        stats_frame = ttk.LabelFrame(main_frame, text="测试结果统计", padding="5")
        stats_frame.pack(fill=tk.X, pady=5)
        self.stats_text = tk.StringVar(value="等待测试...")
        stats_label = ttk.Label(stats_frame, textvariable=self.stats_text, justify=tk.LEFT, font=("TkFixedFont", 9))
        stats_label.pack(anchor=tk.W)

        self.root.protocol("WM_DELETE_WINDOW", self.on_close)

    def log(self, message):
        self.log_text.insert(tk.END, message + "\n")
        self.log_text.see(tk.END)
        self.root.update_idletasks()

    def update_progress(self, elapsed, duration, sent_count):
        percent = int(min(100, elapsed / duration * 100)) if duration > 0 else 0
        self.progress_var.set(percent)
        self.progress_label.config(text=f"{percent}%")
        self.req_count_var.set(f"请求: {sent_count}")

    def finish_test(self, stats):
        self.running = False
        self.start_btn.config(state=tk.NORMAL)
        self.stop_btn.config(state=tk.DISABLED)
        self.progress_var.set(100)
        self.progress_label.config(text="完成")
        self.req_count_var.set(f"请求: {stats['total_requests']}")
        text = (f"总请求数: {stats['total_requests']}\n"
                f"成功数: {stats['successes']}\n"
                f"失败数: {stats['failures']}\n"
                f"成功率: {stats['success_rate']:.2f}%\n"
                f"实际运行时间: {stats['total_time']:.2f} 秒\n"
                f"QPS: {stats['qps']:.2f} req/s\n"
                f"平均响应时间(所有): {stats['avg_time_all']*1000:.2f} ms\n"
                f"平均响应时间(成功): {stats['avg_time_success']*1000:.2f} ms\n"
                f"最小响应时间(成功): {stats['min_time']*1000:.2f} ms\n"
                f"最大响应时间(成功): {stats['max_time']*1000:.2f} ms")
        self.stats_text.set(text)
        self.log("测试结束")

    def start_test(self):
        if self.running:
            return
        url_input = self.url_var.get().strip()
        protocol = self.protocol_var.get()
        url = normalize_url(url_input, protocol)
        if not url:
            self.log("错误：URL 不能为空")
            return
        try:
            concurrency = self.concurrency_var.get()
            duration = self.duration_var.get()
            timeout = self.timeout_var.get()
        except tk.TclError:
            self.log("错误：请检查数值参数格式")
            return
        if duration <= 0:
            self.log("错误：持续时长必须大于0")
            return

        method = self.method_var.get().strip() or "GET"
        codes_str = self.codes_var.get().strip()
        if codes_str:
            try:
                success_codes = [int(c.strip()) for c in codes_str.split(',') if c.strip().isdigit()]
            except ValueError:
                self.log("错误：状态码格式不正确，请用逗号分隔数字")
                return
        else:
            success_codes = [200]
        headers_str = self.headers_var.get().strip()
        headers = {}
        if headers_str:
            for item in headers_str.split(','):
                if ':' in item:
                    key, val = item.split(':', 1)
                    headers[key.strip()] = val.strip()
        data = self.data_var.get().strip() or None
        verify_ssl = not self.ignore_ssl_var.get()
        random_ua = self.random_ua_var.get()
        random_referer = self.random_referer_var.get()

        self.log_text.delete(1.0, tk.END)
        self.stats_text.set("测试进行中...")
        self.progress_var.set(0)
        self.progress_label.config(text="0%")
        self.req_count_var.set("请求: 0")

        self.tester = StressTester(
            url=url,
            concurrency=concurrency,
            duration=duration,
            timeout=timeout,
            method=method,
            success_codes=success_codes,
            headers=headers,
            data=data,
            verify_ssl=verify_ssl,
            random_ua=random_ua,
            random_referer=random_referer
        )
        self.running = True
        self.start_btn.config(state=tk.DISABLED)
        self.stop_btn.config(state=tk.NORMAL)

        self.test_thread = threading.Thread(target=self._run_test, daemon=True)
        self.test_thread.start()
        self.log(f"开始测试: {url} (时长={duration}s, SSL验证={'开启' if verify_ssl else '忽略'}, 随机UA={'开启' if random_ua else '关闭'}, 随机Referer={'开启' if random_referer else '关闭'})")

    def _run_test(self):
        try:
            self.tester.run(
                progress_callback=lambda e, d, s: self.root.after(0, self.update_progress, e, d, s),
                result_callback=lambda stats: self.root.after(0, self.finish_test, stats)
            )
        except Exception as e:
            self.root.after(0, self.log, f"测试异常: {e}")
            self.root.after(0, self.finish_test, {
                'total_requests': 0, 'successes': 0, 'failures': 0,
                'success_rate': 0, 'total_time': 0, 'qps': 0,
                'avg_time_all': 0, 'avg_time_success': 0, 'min_time': 0, 'max_time': 0
            })

    def stop_test(self):
        if self.running and self.tester:
            self.log("正在停止测试...")
            self.tester.set_stop_flag()
            self.stop_btn.config(state=tk.DISABLED)

    def on_close(self):
        if self.running and self.tester:
            self.tester.set_stop_flag()
            if self.test_thread and self.test_thread.is_alive():
                self.test_thread.join(timeout=2.0)
        self.root.destroy()


# ---------- 命令行模式 ----------
def run_cli():
    global stop_flag
    parser = argparse.ArgumentParser(
        description="Web 压力测试工具（持续时长版 + 随机UA/Referer）",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )
    parser.add_argument("-u", "--url", required=True, help="目标 URL (支持 http/https)")
    parser.add_argument("-c", "--concurrency", type=int, default=10, help="并发线程数")
    parser.add_argument("-d", "--duration", type=float, required=True, help="持续测试时长（秒）")
    parser.add_argument("-t", "--timeout", type=float, default=5.0, help="每个请求的超时时间（秒）")
    parser.add_argument("-m", "--method", default="GET", help="HTTP 方法")
    parser.add_argument("--success-codes", type=int, nargs='+', default=[200], help="视为成功的状态码")
    parser.add_argument("--headers", type=str, default=None, help="自定义请求头，格式 'key1:value1,key2:value2'")
    parser.add_argument("--data", type=str, default=None, help="POST 请求体")
    parser.add_argument("--insecure", action="store_true", help="忽略 SSL 证书验证")
    parser.add_argument("--random-ua", dest='random_ua', action='store_true', default=True,
                        help="每次请求使用随机 User-Agent (默认开启)")
    parser.add_argument("--no-random-ua", dest='random_ua', action='store_false',
                        help="禁用随机 User-Agent")
    parser.add_argument("--random-referer", dest='random_referer', action='store_true', default=True,
                        help="每次请求使用随机 Referer (默认开启)")
    parser.add_argument("--no-random-referer", dest='random_referer', action='store_false',
                        help="禁用随机 Referer")
    args = parser.parse_args()

    headers = {}
    if args.headers:
        for item in args.headers.split(','):
            if ':' in item:
                key, val = item.split(':', 1)
                headers[key.strip()] = val.strip()

    signal.signal(signal.SIGINT, signal_handler)

    tester = StressTester(
        url=args.url,
        concurrency=args.concurrency,
        duration=args.duration,
        timeout=args.timeout,
        method=args.method,
        success_codes=args.success_codes,
        headers=headers,
        data=args.data,
        verify_ssl=not args.insecure,
        random_ua=args.random_ua,
        random_referer=args.random_referer
    )

    def progress_cb(elapsed, duration, sent_count):
        percent = min(100, elapsed / duration * 100) if duration > 0 else 0
        print(f"\r进度: {percent:.1f}% | 已发送: {sent_count} 请求 | 已用时间: {elapsed:.1f}s", end='')
        if elapsed >= duration:
            print()

    print(f"目标 URL: {args.url} (SSL验证={'开启' if not args.insecure else '忽略'}, 随机UA={'开启' if args.random_ua else '关闭'}, 随机Referer={'开启' if args.random_referer else '关闭'})")
    print(f"并发数: {args.concurrency}, 持续时长: {args.duration}s")
    print("按 Ctrl+C 提前终止")
    stats = tester.run(progress_callback=progress_cb)

    print("\n========== 测试结果 ==========")
    print(f"总请求数:        {stats['total_requests']}")
    print(f"成功数:          {stats['successes']}")
    print(f"失败数:          {stats['failures']}")
    print(f"成功率:          {stats['success_rate']:.2f}%")
    print(f"实际运行时间:    {stats['total_time']:.2f} 秒")
    print(f"QPS:             {stats['qps']:.2f} req/s")
    print(f"平均响应时间(所有请求): {stats['avg_time_all']*1000:.2f} ms")
    if stats['successes'] > 0:
        print(f"平均响应时间(成功请求): {stats['avg_time_success']*1000:.2f} ms")
        print(f"最小响应时间(成功请求): {stats['min_time']*1000:.2f} ms")
        print(f"最大响应时间(成功请求): {stats['max_time']*1000:.2f} ms")
    print("==============================")


# ---------- 主入口 ----------
def main():
    if len(sys.argv) > 1:
        run_cli()
    else:
        if not GUI_AVAILABLE:
            print("错误：tkinter 未安装，无法启动 GUI。请安装 tkinter 或使用命令行模式。")
            sys.exit(1)
        root = tk.Tk()
        app = StressTestApp(root)
        root.mainloop()


if __name__ == "__main__":
    main()

声明：本次代码通过向Web服务器发送大量的请求，其行为与DoS攻击无异，切勿对他人网站进行测试。